sanguo/sanguo_vnpy_v2
2
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(data): 恢复双源(task#79)—撮合raw+策略qfq, 分红除权准确

Browse Source 
用户要模拟=回测准确: raw除权缺口致MA假信号, 必须双源。
- data_source: qfq→qfq_dir(干净qfq), raw→raw_dir; _check_adjust_cfg(cfg提供才校验)
- engine 双bar流: step(raw_bars,qfq_bars)撮合/盯市raw+策略on_bar qfq; run zip(raw,qfq)
- live_orchestrator: warmup用qfq(信号am); 去adjust参数(双源固定)
- raw_redownload --adjust(''raw/'qfq'); config qfq_dir
- 113/113通过
This commit is contained in:
Claude Dev
2026-07-08 07:21:33 +08:00
parent ab703e93ba
commit c6b19f4244
7 changed files with 55 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/data_platform.yaml
+1
View File
@@ -2,6 +2,7 @@
data_paths:
daily_dir: /volume1/stock/A股数据/日线数据/daily
raw_dir: /volume1/stock/A股数据/日线数据/raw
qfq_dir: /volume1/stock/A股数据/日线数据/qfq
minute_15_dir: /volume1/stock/minute_kline/15min
vnpy_db: /volume1/stock/sanguo_vnpy/data/quant_trading.db
stock_list: /volume1/stock/A股数据/stock_info/stock_basic_info_raw_20260326_113530.csv
sanguo_trader/data_source.py
+16 -12
View File
@@ -1,8 +1,9 @@
"""模拟盘行情统一接口（raw 真实价 / qfq 前复权，spec §3.3 / §5）。
"""模拟盘行情统一接口（raw 真实价 / qfq 前复权 双源，spec §3.3 / §5）。
raw：撮合/涨跌停/成交价用真实价（adjustflag=3 / akshare adjust=""），
从 cfg.data_paths['raw_dir'] 读 parquet（raw 重下脚本生成，单一 adjust 不混源）。
qfq：策略信号用前复权（无除权缺口），从 cfg.data_paths['daily_dir'] 读。
双源（分红除权准确方案）：
- raw：撮合/涨跌停/成交价用真实价（adjustflag=3 / akshare adjust=""），cfg.data_paths['raw_dir']
- qfq：策略信号用前复权（无除权缺口，MA 信号准），cfg.data_paths['qfq_dir']
- daily_dir 是历史 mixed（hfq+raw），仅 backtest 兼容，模拟盘不用
_read_fn 内 lazy import datareader，避免模块级依赖 vnpy 链（tzlocal 等），
本机无 vnpy 完整依赖时仍可 import + 单测（mock _read_fn）。
@@ -27,7 +28,7 @@ def _read_fn(interval: str):
def _resolve_dir_key(adjust: str, interval: str) -> str:
"""adjust → cfg.data_paths 的目录 key。
raw 仅支持日线（raw 15min 待分红除权分期项）；qfq/默认按 interval 选日线/15min。
raw 仅日线（raw 15min 待分期项）；qfq/默认按 interval 选。
"""
if adjust == "raw":
if interval != "d":
@@ -36,15 +37,18 @@ def _resolve_dir_key(adjust: str, interval: str) -> str:
"（raw 15min 待分期项）"
)
return "raw_dir"
if adjust == "qfq":
return "qfq_dir" # 干净 qfq（daily_dir 是 mixed，仅 backtest 兼容）
return "daily_dir" if interval == "d" else "minute_15_dir"
def _check_raw_cfg(adjust: str, cfg) -> None:
"""raw 模式需 raw_dir 配置，缺失明确报错（不静默 fallback 到 qfq，避免混源）。"""
if adjust == "raw" and (not cfg or "raw_dir" not in getattr(cfg, "data_paths", {})):
def _check_adjust_cfg(adjust: str, cfg) -> None:
"""raw/qfq 需对应 dir 配置，缺失明确报错（不静默 fallback，避免混源）。"""
need = {"raw": "raw_dir", "qfq": "qfq_dir"}.get(adjust)
if need and cfg and need not in getattr(cfg, "data_paths", {}):
raise ValueError(
"raw 模式需 cfg.data_paths['raw_dir']（未配置；"
"先用 scripts/data_platform/raw_redownload.py 生成 raw parquet）"
f"{adjust} 模式需 cfg.data_paths['{need}']（未配置；"
f"先用 raw_redownload.py --adjust {adjust} 生成 parquet）"
)
@@ -58,7 +62,7 @@ def iter_bars(
) -> Iterator[tuple]:
"""按日期 cross-section yield (date, {symbol: BarData})。"""
dir_key = _resolve_dir_key(adjust, interval)
_check_raw_cfg(adjust, cfg)
_check_adjust_cfg(adjust, cfg)
read_fn = _read_fn(interval)
by_date: dict = {}
for sym in symbols:
@@ -74,7 +78,7 @@ def fetch_day(symbol: str, date: str, interval: str,
adjust: str = "qfq", cfg=None):
"""实走模式拉当日 bar（C-S3 用）。"""
dir_key = _resolve_dir_key(adjust, interval)
_check_raw_cfg(adjust, cfg)
_check_adjust_cfg(adjust, cfg)
read_fn = _read_fn(interval)
bars = read_fn(symbol, date, date, cfg, dir_key)
return bars[-1] if bars else None
sanguo_trader/engine.py
+27 -23
View File
@@ -1,10 +1,9 @@
"""PaperEngine 模拟盘主循环（逐根 bar 重放 + 双层记账 + 持久化，spec §4/§9）。
run()：逐 bar → T+1 解冻 → 撮合上一根 next_open pending（用当前 bar）→ 喂策略
on_bar 收新单 → current_close 当根撮合 / next_open 缓冲到下根 → 盯市 → 入库。
adjust 默认 raw（真实价）：撮合/涨跌停/成交价/信号共用一套真实价 bar。
除权缺口对 MA 信号的影响留分期项（分红除权）处理。
双源（分红除权准确方案，task #79 恢复）：
- 撮合/涨跌停/盯市用 **raw**（真实价，涨跌停/成交真实）
- 策略 on_bar 信号用 **qfq**（前复权，无除权缺口 → MA 信号准）
run() 双迭代器 zip(raw, qfq) 同日期对齐；step(raw_bars, qfq_bars)。
"""
import logging
@@ -41,7 +40,7 @@ class PaperEngine:
def __init__(self, account: Account, runners: list[StrategyRunner],
data_source, cfg, db_path: str, account_id: int,
symbols: list[str], start: str, end: str,
interval: str = "d", adjust: str = "raw") -> None:
interval: str = "d") -> None:
self.account = account
self.runners = runners
self.data_source = data_source
@@ -52,34 +51,34 @@ class PaperEngine:
self.start = start
self.end = end
self.interval = interval
self.adjust = adjust
def step(self, bar_date, bars, prev_close, pending):
"""单根 bar 推进（回放 run 循环调；C-S3 实走 scheduler 每日调）。
def step(self, bar_date, raw_bars, qfq_bars, prev_close, pending):
"""单根 bar 推进（回放 run 循环调；实走 live_step 调）。
返回 (新 pending, 当根 closes)——实走每日喂当日 bar 调一次。
撮合/盯市用 raw_bars（真实价）；策略 on_bar 用 qfq_bars（信号准）。
返回 (新 pending, 当根 closes)。
"""
self._bar_count = getattr(self, "_bar_count", 0) + 1
self.account.unfreeze_all()
for r in self.runners:
r.unfreeze_all()
# 1. 撮合上一根 pending（next_open，用当前 bar）
# 1. 撮合上一根 pending（next_open，用当日 raw bar）
if pending:
for order, runner in pending:
self._match(order, runner, bars, prev_close, bar_date)
self._match(order, runner, raw_bars, prev_close, bar_date)
pending = []
# 2. 喂策略 on_bar → 收新单
# 2. 喂策略 on_bar（qfq 信号）→ 收新单 → 当根撮合 raw / 缓冲 next_open
for runner in self.runners:
sym = runner.symbol
if sym and sym in bars:
runner.paper_cta_engine.on_bar(bars[sym])
if sym and sym in qfq_bars:
runner.paper_cta_engine.on_bar(qfq_bars[sym])
for order in runner.paper_cta_engine.pop_orders():
if order.match_session == MatchSession.NEXT_OPEN:
pending.append((order, runner))
else: # current_close 当根撮合
self._match(order, runner, bars, prev_close, bar_date)
# 3. 盯市 + 入库
closes = {s: bars[s].close_price for s in bars}
else: # current_close 当根撮合（raw）
self._match(order, runner, raw_bars, prev_close, bar_date)
# 3. 盯市 raw + 入库
closes = {s: raw_bars[s].close_price for s in raw_bars}
self.account.mark_to_market(closes)
save_daily_balance(
self.db_path, self.account_id, str(bar_date),
@@ -90,12 +89,17 @@ class PaperEngine:
return pending, closes
def run(self) -> None:
"""双源 zip(raw, qfq) 同日期对齐，逐根 step。"""
prev_close: dict[str, float] = {}
pending: list = [] # [(order, runner)] next_open 待下根撮合
for bar_date, bars in self.data_source.iter_bars(
self.symbols, self.start, self.end, self.interval, self.adjust, None
):
pending, closes = self.step(bar_date, bars, prev_close, pending)
raw_iter = self.data_source.iter_bars(
self.symbols, self.start, self.end, self.interval, "raw", None
)
qfq_iter = self.data_source.iter_bars(
self.symbols, self.start, self.end, self.interval, "qfq", None
)
for (rdate, raw_bars), (_qdate, qfq_bars) in zip(raw_iter, qfq_iter):
pending, closes = self.step(rdate, raw_bars, qfq_bars, prev_close, pending)
prev_close = closes
def _match(self, order, runner, bars, prev_close, bar_date) -> None:
sanguo_trader/live_orchestrator.py
+2 -2
View File
@@ -97,7 +97,7 @@ def live_step(db_path: str, account_id: int, data_source, cfg, today: str | None
start_date = acc.get("start_date") or today
if start_date < yesterday:
for _wd, wbars in data_source.iter_bars(
symbols, start_date, yesterday, interval, adjust="raw", cfg=cfg
symbols, start_date, yesterday, interval, adjust="qfq", cfg=cfg
):
for runner in runners:
if runner.symbol in wbars:
@@ -137,7 +137,7 @@ def live_step(db_path: str, account_id: int, data_source, cfg, today: str | None
transfer_fee_rate=acc["transfer_fee_rate"], min_commission=acc["min_commission"],
)
pe = PaperEngine(account, runners, data_source, acc_cfg, db_path, account_id,
symbols, acc.get("start_date") or today, today, interval, adjust="raw")
symbols, acc.get("start_date") or today, today, interval)
pending_new, _closes = pe.step(today, bars, prev_close, pending)
# 6. 存状态（pending + positions）
scripts/data_platform/raw_redownload.py
+5 -4
View File
@@ -48,15 +48,15 @@ def prefix_for(code: str) -> str:
return "sh" if code.startswith(("60", "68", "51", "56", "58")) else "sz"
def download_one(ak, code: str, start: str, end: str):
"""新浪源拉 raw 日线，返回 (df, None) 或 (None, err)。"""
def download_one(ak, code: str, start: str, end: str, adjust: str = ""):
"""新浪源拉日线（adjust="" raw / "qfq" 前复权），返回 (df, None) 或 (None, err)。"""
sym = f"{prefix_for(code)}{code}"
try:
df = ak.stock_zh_a_daily(
symbol=sym,
start_date=start.replace("-", ""),
end_date=end.replace("-", ""),
adjust="", # raw 不复权
adjust=adjust,
)
except Exception as e: # noqa: BLE001
return None, f"{type(e).__name__}: {str(e)[:100]}"
@@ -100,6 +100,7 @@ def main():
ap.add_argument("--all", action="store_true", help="全市场（读 STOCK_LIST csv）")
ap.add_argument("--start", default="2024-01-01")
ap.add_argument("--end", default=None, help="默认今天")
ap.add_argument("--adjust", default="", help="复权: '' raw / 'qfq' 前复权（双源用）")
args = ap.parse_args()
end = args.end or time.strftime("%Y-%m-%d")
@@ -122,7 +123,7 @@ def main():
ok = fail = rows = 0
for i, code in enumerate(codes, 1):
df, err = download_one(ak, code, args.start, end)
df, err = download_one(ak, code, args.start, end, args.adjust)
if df is None:
fail += 1
log.warning("[%d/%d] %s FAIL %s", i, len(codes), code, err)
tests/trader/test_data_source.py
+2 -2
View File
@@ -34,7 +34,7 @@ def test_iter_bars_cross_section_multi_symbol(monkeypatch):
assert len(sections) == 1
_date, d = sections[0]
assert d["600000"].close_price == 10.5
assert all(k == "daily_dir" for k in seen) # 默认 qfq → daily_dir
assert all(k == "qfq_dir" for k in seen) # 默认 qfq → qfq_dir（双源）
def test_iter_bars_raw_uses_raw_dir(monkeypatch, caplog):
@@ -90,4 +90,4 @@ def test_fetch_day_returns_last_bar(monkeypatch):
monkeypatch.setattr("sanguo_trader.data_source._read_fn", lambda iv: mock_read)
bar = fetch_day("600000", "2024-01-02", "d")
assert bar.close_price == 10.5 # 取最后一个
assert seen == ["daily_dir"]
assert seen == ["qfq_dir"] # 默认 qfq → qfq_dir（双源）
tests/trader/test_engine.py
+2 -2
View File
@@ -101,12 +101,12 @@ def test_engine_step_single_bar_advances(tmp_path):
]
pe, db, aid, account, runner = _build(tmp_path, sections)
# day1 step：_AlwaysBuyStrategy 买单（NEXT_OPEN）→ 进 pending，当根不撮合
pending, closes = pe.step("2024-01-01", sections[0][1], {}, [])
pending, closes = pe.step("2024-01-01", sections[0][1], sections[0][1], {}, [])
assert len(pending) == 1
assert account.positions.get("600000") is None
assert closes["600000"] == 10.0
# day2 step：撮合 day1 pending @ open 10.5；策略 on_bar(day2) 又发单进 pending 等 day3
pending2, closes2 = pe.step("2024-01-02", sections[1][1], closes, pending)
pending2, closes2 = pe.step("2024-01-02", sections[1][1], sections[1][1], closes, pending)
assert len(pending2) == 1 # day2 新信号（无 day3 不撮合）
assert account.positions["600000"].volume == 100 # day1 单 day2 open 10.5 撮合 100 股
# step 入库（day1+day2 各一条余额）