diff --git a/sanguo_trader/engine.py b/sanguo_trader/engine.py new file mode 100644 index 0000000..398a500 --- /dev/null +++ b/sanguo_trader/engine.py @@ -0,0 +1,112 @@ +"""PaperEngine 模拟盘主循环(逐根 bar 重放 + 双层记账 + 持久化,spec §4/§9)。 + +run():逐 bar → T+1 解冻 → 撮合上一根 next_open pending(用当前 bar)→ 喂策略 +on_bar 收新单 → current_close 当根撮合 / next_open 缓冲到下根 → 盯市 → 入库。 +raw 首版 fallback qfq(spec §17),信号与撮合共用一套(标注)。 +""" +import logging + +import pandas as pd + +from .account import Account +from .matcher import cross_order +from .models import MatchSession, 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") -> 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 + + def run(self) -> None: + prev_close: dict[str, float] = {} + pending: list = [] # [(order, runner)] next_open 待下根撮合 + bar_count = 0 + for bar_date, bars in self.data_source.iter_bars( + self.symbols, self.start, self.end, self.interval, "qfq", None + ): + bar_count += 1 + self.account.unfreeze_all() + for r in self.runners: + r.unfreeze_all() + # 1. 撮合上一根 pending(next_open,用当前 bar) + if pending: + for order, runner in pending: + self._match(order, runner, bars, prev_close, bar_date) + pending = [] + # 2. 喂策略 on_bar → 收新单 + for runner in self.runners: + sym = runner.symbol + if sym and sym in bars: + runner.paper_cta_engine.on_bar(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} + self.account.mark_to_market(closes) + save_daily_balance( + self.db_path, self.account_id, str(bar_date), + self.account.cash, self.account.market_value, self.account.equity, + is_checkpoint=(bar_count % 500 == 0), + ) + update_checkpoint(self.db_path, self.account_id, str(bar_date)) + prev_close = closes + + 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 self.account.cash_enough(result): + 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) diff --git a/sanguo_trader/strategy_runner.py b/sanguo_trader/strategy_runner.py index a111aab..906c6ad 100644 --- a/sanguo_trader/strategy_runner.py +++ b/sanguo_trader/strategy_runner.py @@ -7,10 +7,12 @@ from .position_ledger import PositionLedger class StrategyRunner: - def __init__(self, strategy_id: str, strategy=None, paper_cta_engine=None) -> None: + def __init__(self, strategy_id: str, strategy=None, paper_cta_engine=None, + symbol: str = "") -> None: self.strategy_id: str = strategy_id self.strategy = strategy self.paper_cta_engine = paper_cta_engine + self.symbol: str = symbol self.positions: dict[str, PositionLedger] = {} self.realized_pnl: float = 0.0 self.commission_paid: float = 0.0 diff --git a/tests/trader/test_engine.py b/tests/trader/test_engine.py new file mode 100644 index 0000000..b6e168a --- /dev/null +++ b/tests/trader/test_engine.py @@ -0,0 +1,93 @@ +"""PaperEngine 主循环测试(mock data_source + mock 策略,spec §4)。""" +from types import SimpleNamespace + +from sanguo_trader.account import Account +from sanguo_trader.cta_adapter import PaperCtaEngine +from sanguo_trader.engine import PaperEngine +from sanguo_trader.models import AccountConfig, MatchSession +from sanguo_trader.persistence import ( + init_db, list_daily_balance, list_trades, save_account, +) +from sanguo_trader.strategy_runner import StrategyRunner + + +def _bar(date, o, h, l, c): + return SimpleNamespace(open_price=o, high_price=h, low_price=l, + close_price=c, datetime=date) + + +class _FakeDataSource: + def __init__(self, sections): + self.sections = sections + + def iter_bars(self, symbols, start, end, interval, adjust="qfq", cfg=None): + del symbols, start, end, interval, adjust, cfg + for date, bars in self.sections: + yield date, bars + + +class _AlwaysBuyStrategy: + def __init__(self, engine, vt_symbol): + self.cta_engine = engine + self.vt_symbol = vt_symbol + + def on_bar(self, bar): + self.cta_engine.send_order(self, "LONG", "OPEN", bar.close_price, 100) + + +def _build(tmp_path, sections, match_session=MatchSession.NEXT_OPEN): + db = str(tmp_path / "e.db") + init_db(db) + aid = save_account(db, {"name": "t", "initial_capital": 1_000_000}) + cfg = AccountConfig(initial_capital=1_000_000) + account = Account(1_000_000) + cta = PaperCtaEngine("s1", match_session=match_session) + strat = _AlwaysBuyStrategy(cta, "600000.SSE") + cta.set_strategy(strat) + runner = StrategyRunner("s1", strategy=strat, paper_cta_engine=cta, symbol="600000") + pe = PaperEngine(account, [runner], _FakeDataSource(sections), cfg, db, aid, + symbols=["600000"], start="2024-01-01", end="2024-12-31") + return pe, db, aid, account, runner + + +def test_engine_next_open_fills_at_next_bar_open(tmp_path): + sections = [ + ("2024-01-01", {"600000": _bar("2024-01-01", 10.0, 10.5, 9.5, 10.0)}), + ("2024-01-02", {"600000": _bar("2024-01-02", 10.5, 11.0, 10.0, 10.8)}), + ("2024-01-03", {"600000": _bar("2024-01-03", 10.8, 11.5, 10.5, 11.0)}), + ] + pe, db, aid, *_ = _build(tmp_path, sections) + pe.run() + fills = [t for t in list_trades(db, aid) if not t["rejected"]] + # day1 信号→day2 撮合@10.5(open);day2 信号→day3 撮合@10.8(open);day3 信号无day4 + assert len(fills) == 2 + assert fills[0]["price"] == 10.5 + assert fills[1]["price"] == 10.8 + + +def test_engine_current_close_fills_same_bar(tmp_path): + sections = [ + ("2024-01-01", {"600000": _bar("2024-01-01", 10.0, 10.5, 9.5, 10.0)}), + ("2024-01-02", {"600000": _bar("2024-01-02", 10.5, 11.0, 10.0, 10.8)}), + ] + pe, db, aid, *_ = _build(tmp_path, sections, match_session=MatchSession.CURRENT_CLOSE) + pe.run() + fills = [t for t in list_trades(db, aid) if not t["rejected"]] + assert len(fills) == 2 + assert fills[0]["price"] == 10.0 # 当根 close + assert fills[1]["price"] == 10.8 + + +def test_engine_daily_balance_and_consistency(tmp_path): + sections = [ + ("2024-01-01", {"600000": _bar("2024-01-01", 10.0, 10.5, 9.5, 10.0)}), + ("2024-01-02", {"600000": _bar("2024-01-02", 10.5, 11.0, 10.0, 10.8)}), + ("2024-01-03", {"600000": _bar("2024-01-03", 10.8, 11.5, 10.5, 11.0)}), + ] + pe, db, aid, account, runner = _build(tmp_path, sections) + pe.run() + balances = list_daily_balance(db, aid) + assert len(balances) == 3 + # 总账持仓 = 分户持仓(day2+day3 各买100 = 200) + assert account.positions["600000"].volume == 200 + assert runner.positions["600000"].volume == 200