From c6b19f42445e08b5c9c65c0da5383c3b9e8cd60d Mon Sep 17 00:00:00 2001 From: claude_dev Date: Wed, 8 Jul 2026 07:21:33 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20=E6=81=A2=E5=A4=8D=E5=8F=8C?= =?UTF-8?q?=E6=BA=90(task#79)=E2=80=94=E6=92=AE=E5=90=88raw+=E7=AD=96?= =?UTF-8?q?=E7=95=A5qfq,=20=E5=88=86=E7=BA=A2=E9=99=A4=E6=9D=83=E5=87=86?= =?UTF-8?q?=E7=A1=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户要模拟=回测准确: 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通过 --- config/data_platform.yaml | 1 + sanguo_trader/data_source.py | 28 ++++++++------ sanguo_trader/engine.py | 50 +++++++++++++------------ sanguo_trader/live_orchestrator.py | 4 +- scripts/data_platform/raw_redownload.py | 9 +++-- tests/trader/test_data_source.py | 4 +- tests/trader/test_engine.py | 4 +- 7 files changed, 55 insertions(+), 45 deletions(-) diff --git a/config/data_platform.yaml b/config/data_platform.yaml index 75ef439..d6ea536 100644 --- a/config/data_platform.yaml +++ b/config/data_platform.yaml @@ -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 diff --git a/sanguo_trader/data_source.py b/sanguo_trader/data_source.py index 43721e0..01c0b25 100644 --- a/sanguo_trader/data_source.py +++ b/sanguo_trader/data_source.py @@ -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 diff --git a/sanguo_trader/engine.py b/sanguo_trader/engine.py index 2535c78..fcf05e4 100644 --- a/sanguo_trader/engine.py +++ b/sanguo_trader/engine.py @@ -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: diff --git a/sanguo_trader/live_orchestrator.py b/sanguo_trader/live_orchestrator.py index 5943fc0..dab2436 100644 --- a/sanguo_trader/live_orchestrator.py +++ b/sanguo_trader/live_orchestrator.py @@ -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） diff --git a/scripts/data_platform/raw_redownload.py b/scripts/data_platform/raw_redownload.py index 328e0b4..eb6a13e 100644 --- a/scripts/data_platform/raw_redownload.py +++ b/scripts/data_platform/raw_redownload.py @@ -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) diff --git a/tests/trader/test_data_source.py b/tests/trader/test_data_source.py index fd4380b..1a1e853 100644 --- a/tests/trader/test_data_source.py +++ b/tests/trader/test_data_source.py @@ -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（双源） diff --git a/tests/trader/test_engine.py b/tests/trader/test_engine.py index e48ef3c..bf9aa04 100644 --- a/tests/trader/test_engine.py +++ b/tests/trader/test_engine.py @@ -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 各一条余额）