"""PaperEngine 模拟盘主循环（逐根 bar 重放 + 双层记账 + 持久化，spec §4/§9）。 双源（分红除权准确方案，task #79 恢复）： - 撮合/涨跌停/盯市用 **raw**（真实价，涨跌停/成交真实） - 策略 on_bar 信号用 **qfq**（前复权，无除权缺口 → MA 信号准） run() 双迭代器 zip(raw, qfq) 同日期对齐；step(raw_bars, qfq_bars)。 """ import logging import pandas as pd from .account import Account from .matcher import cross_order from .models import MatchSession, OrderSide, PaperTrade from .persistence import save_daily_balance, save_trade, update_checkpoint from .strategy_runner import StrategyRunner logger = logging.getLogger(__name__) def _to_series(bar) -> pd.Series: return pd.Series({ "open": bar.open_price, "high": bar.high_price, "low": bar.low_price, "close": bar.close_price, "date": getattr(bar, "datetime", ""), }) def _trade_to_dict(t: PaperTrade, bar_date) -> dict: return { "strategy_id": t.strategy_id, "symbol": t.symbol, "direction": t.side.value, "price": t.price, "volume": t.volume, "commission": t.commission, "stamp_duty": t.stamp_duty, "transfer_fee": t.transfer_fee, "bar_date": str(bar_date), "match_session": t.match_session.value, } 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", risk_free_rate: float = 0.0, dividends_by_date: dict | None = None) -> None: self.account = account self.runners = runners self.data_source = data_source self.cfg = cfg self.db_path = db_path self.account_id = account_id self.symbols = symbols self.start = start self.end = end self.interval = interval self.risk_free_rate = risk_free_rate # spec §195 资金占用成本归因 # spec §295 分红送股日历 {ex_date_str: {symbol: DividendEvent}}（缺省空=不处理） self.dividends_by_date = dividends_by_date or {} def step(self, bar_date, raw_bars, qfq_bars, prev_close, pending): """单根 bar 推进（回放 run 循环调；实走 live_step 调）。 撮合/盯市用 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() # 0. 除权除息日：分红送股调整（开盘前持仓享权，spec §295） self._apply_dividends(bar_date) # 1. 撮合上一根 pending（next_open，用当日 raw bar） if pending: for order, runner in pending: self._match(order, runner, raw_bars, prev_close, bar_date) pending = [] # 2. 喂策略 on_bar（qfq 信号）→ 收新单 → 当根撮合 raw / 缓冲 next_open for runner in self.runners: sym = runner.symbol 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 当根撮合（raw） self._match(order, runner, raw_bars, prev_close, bar_date) # 3. 盯市 raw + 入库（停牌缺 bar 用 prev_close 兜底，spec §295） closes = {s: raw_bars[s].close_price for s in raw_bars} self.account.mark_to_market(closes, prev_close) # 资金占用成本归因（spec §195）：每策略持仓按无风险利率日扣，仅记 per_strategy_pnl # 不扣 account.cash —— 总账净值保持真实，占用成本是分策略展示用归因 per_strategy_pnl: dict = {} if self.risk_free_rate > 0: for r in self.runners: cost = r.daily_borrow_cost(closes, self.risk_free_rate) per_strategy_pnl[r.strategy_id] = {"borrow_cost": cost} save_daily_balance( self.db_path, self.account_id, str(bar_date), self.account.cash, self.account.market_value, self.account.equity, per_strategy_pnl=per_strategy_pnl or None, is_checkpoint=(self._bar_count % 500 == 0), ) update_checkpoint(self.db_path, self.account_id, str(bar_date)) return pending, closes def run(self) -> None: """双源 zip(raw, qfq) 同日期对齐，逐根 step。 分红送股日历由调用方经 dividends_by_date 注入（见 __init__）； 回测脚本可用 sanguo_data.dividend_source.build_dividend_calendar 预拉。 """ prev_close: dict[str, float] = {} pending: list = [] # [(order, runner)] next_open 待下根撮合 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 _apply_dividends(self, bar_date) -> None: """除权除息日调整持仓（spec §295）：现金分红按除权前持仓量，再 apply_split。 双层记账一致性：Account 与各 StrategyRunner 分户同步 split。 """ events = self.dividends_by_date.get(str(bar_date)) if not events: return for sym, ev in events.items(): pos = self.account.positions.get(sym) if pos is None or pos.volume <= 0: continue # 现金分红用除权前持仓量（apply_split 前的 volume） self.account.apply_cash_dividend(sym, ev.cash_per_share) pos.apply_split(ev.split_factor) # 分户同步 split（保持总账=分户之和） for r in self.runners: rpos = r.positions.get(sym) if rpos is not None and rpos.volume > 0: rpos.apply_split(ev.split_factor) def _match(self, order, runner, bars, prev_close, bar_date) -> None: if order.symbol not in bars: return match_bar = _to_series(bars[order.symbol]) pc = prev_close.get(order.symbol, order.price) result = cross_order(order, match_bar, pc, self.cfg) if isinstance(result, PaperTrade): if result.side == OrderSide.SELL: # A 股不能做空：SELL 超过可卖持仓 → 拒单（不开空仓） pos = self.account.positions.get(order.symbol) if pos is None or pos.available < result.volume: save_trade(self.db_path, self.account_id, {"strategy_id": order.strategy_id, "symbol": order.symbol, "bar_date": str(bar_date)}, rejected=True, reject_reason="insufficient_position_no_short") return if self.account.cash_enough(result): # 软限额：BUY 检查该策略已用额度（消除多策略撮合顺序依赖，spec §195） if result.side == OrderSide.BUY: closes = {sym: b.close_price for sym, b in bars.items()} if (runner.used_allocation(closes) + result.price * result.volume > runner.max_allocation): save_trade(self.db_path, self.account_id, {"strategy_id": order.strategy_id, "symbol": order.symbol, "bar_date": str(bar_date)}, rejected=True, reject_reason="max_allocation_exceeded") return self.account.apply_trade(result) runner.apply_trade(result) save_trade(self.db_path, self.account_id, _trade_to_dict(result, bar_date)) else: save_trade(self.db_path, self.account_id, {"strategy_id": order.strategy_id, "symbol": order.symbol, "bar_date": str(bar_date)}, rejected=True, reject_reason="insufficient_cash") else: # PaperReject save_trade(self.db_path, self.account_id, {"strategy_id": order.strategy_id, "symbol": order.symbol, "bar_date": str(bar_date)}, rejected=True, reject_reason=result.reason)