diff --git a/src/quant_engine/execution.py b/src/quant_engine/execution.py index 279c030..b26d4c2 100644 --- a/src/quant_engine/execution.py +++ b/src/quant_engine/execution.py @@ -19,6 +19,7 @@ from __future__ import annotations +import math from collections.abc import Mapping from dataclasses import dataclass from typing import Any @@ -344,19 +345,90 @@ class DailyExecution: """单日执行记录。""" date: str - executions: list[ExecutionResult] + executions: tuple[ExecutionResult, ...] nav_before: float nav_after: float rebalance_triggered: bool -def simulate_multi_day( +@dataclass(frozen=True) +class ExecutionSimulationResult: + """单次多日仿真的持仓与执行审计结果。""" + + initial_cash: float + positions: tuple[DailyPosition, ...] + daily_executions: tuple[DailyExecution, ...] + + @property + def nav_series(self) -> pd.Series: + """返回按日期索引的日末 NAV 副本。""" + return pd.Series( + [position.portfolio_value for position in self.positions], + index=[position.date for position in self.positions], + dtype=float, + ) + + @property + def total_costs(self) -> float: + """汇总实际成交产生的成本。""" + return sum( + execution.total_cost + for daily in self.daily_executions + for execution in daily.executions + ) + + @property + def total_turnover(self) -> float: + """汇总实际成交金额。""" + return sum( + execution.executed_value + for daily in self.daily_executions + for execution in daily.executions + ) + + @property + def total_rebalances(self) -> int: + """返回至少有一笔实际成交的调仓日数量。""" + return sum(daily.rebalance_triggered for daily in self.daily_executions) + + @property + def final_portfolio_value(self) -> float: + """返回最后一个日末 NAV；空输入时返回初始资金。""" + if not self.positions: + return self.initial_cash + return self.positions[-1].portfolio_value + + @property + def return_pct(self) -> float: + """返回相对初始资金的百分比收益。""" + if self.initial_cash == 0: + return 0.0 + return (self.final_portfolio_value / self.initial_cash - 1.0) * 100.0 + + +def _blocked_execution(stock_code: str, target_value: float, reason: str) -> ExecutionResult: + """构造未成交但可审计的执行记录。""" + return ExecutionResult( + stock_code=stock_code, + target_value=target_value, + executed_value=0.0, + commission=0.0, + stamp_tax=0.0, + slippage_cost=0.0, + total_cost=0.0, + net_cash_flow=0.0, + partial_fill_pct=0.0, + blocked_reason=reason, + ) + + +def simulate_multi_day_with_audit( target_weights_history: list[tuple[str, dict[str, float]]], price_history: list[tuple[str, dict[str, float]]], initial_cash: float, config: ExecutionConfig | None = None, -) -> list[DailyPosition]: - """多日组合仿真（NAV 序列）。 +) -> ExecutionSimulationResult: + """按目标权重差额推进组合，并返回唯一事实来源的审计结果。 Args: target_weights_history: [(date, {stock_code: target_weight})] @@ -365,97 +437,115 @@ def simulate_multi_day( config: 执行配置 Returns: - DailyPosition 列表（每日 NAV 快照）。 + 日末持仓快照与逐日成交记录组成的不可变结果。 Note: - 调仓频率 = target_weights_history 的频率（每日 / 每周 / 每月都行） - - 每日先按当日 close 估值，再按当日 target 调仓（下一交易日生效） - - 此处简化：调仓使用当日 close 价格 + - 每日先按当日 close 估值，再交易“目标市值 - 当前市值”的差额 + - 此处简化为当日 close 成交；调用方必须传入已正确滞后的目标权重 """ if config is None: config = ExecutionConfig() + if not math.isfinite(initial_cash) or initial_cash < 0: + raise ValueError(f"initial_cash must be finite and non-negative, got {initial_cash}") if len(target_weights_history) != len(price_history): raise ValueError("target_weights_history and price_history must have same length") if not target_weights_history: - return [] + return ExecutionSimulationResult(initial_cash, (), ()) + cash = initial_cash holdings: dict[str, float] = {} positions: list[DailyPosition] = [] - for (date, targets), (_, prices) in zip(target_weights_history, price_history, strict=True): - # 1) 先按当日收盘价估值 - portfolio_value = cash + sum( - shares * prices.get(code, 0.0) for code, shares in holdings.items() + daily_executions: list[DailyExecution] = [] + + for (date, targets), (price_date, prices) in zip( + target_weights_history, price_history, strict=True + ): + if date != price_date: + raise ValueError( + f"target and price dates must match, got {date!r} and {price_date!r}" + ) + + nav_before = cash + sum( + shares * prices.get(stock_code, 0.0) + for stock_code, shares in holdings.items() ) - positions.append( - DailyPosition( + effective_targets = dict.fromkeys(holdings, 0.0) + effective_targets.update(targets) + trade_weights: dict[str, float] = {} + rejected: list[ExecutionResult] = [] + + for stock_code, target_weight in effective_targets.items(): + price = prices.get(stock_code) + target_value = float(target_weight) * nav_before + if price is None or not math.isfinite(price) or price <= 0: + if target_value != 0 or holdings.get(stock_code, 0.0) != 0: + rejected.append(_blocked_execution(stock_code, target_value, "missing_price")) + continue + + current_value = holdings.get(stock_code, 0.0) * price + trade_value = target_value - current_value + if abs(trade_value) < config.min_trade_amount or math.isclose( + trade_value, 0.0, abs_tol=1e-12 + ): + continue + if nav_before == 0: + rejected.append(_blocked_execution(stock_code, trade_value, "zero_nav")) + continue + trade_weights[stock_code] = trade_value / nav_before + + filled = simulate_execution(trade_weights, nav_before, config) + for execution in filled: + price = prices[execution.stock_code] + share_change = abs(execution.target_value) / price + if execution.target_value > 0: + holdings[execution.stock_code] = ( + holdings.get(execution.stock_code, 0.0) + share_change + ) + else: + held = holdings.get(execution.stock_code, 0.0) + holdings[execution.stock_code] = max(0.0, held - share_change) + if holdings[execution.stock_code] < 1e-6: + del holdings[execution.stock_code] + cash += execution.net_cash_flow + + executions = (*filled, *rejected) + nav_after = cash + sum( + shares * prices.get(stock_code, 0.0) + for stock_code, shares in holdings.items() + ) + positions.append(DailyPosition(date, cash, dict(holdings), nav_after)) + daily_executions.append( + DailyExecution( date=date, - cash=cash, - holdings=dict(holdings), - portfolio_value=portfolio_value, + executions=executions, + nav_before=nav_before, + nav_after=nav_after, + rebalance_triggered=bool(filled), ) ) - # 2) 计算 effective_targets（包含需要平仓的零权重） - effective_targets: dict[str, float] = dict(targets) - for held_code in holdings: - if held_code not in effective_targets: - effective_targets[held_code] = 0.0 - # 3) 调仓（只对非零目标调用 simulate_execution） - non_zero_targets = {k: v for k, v in effective_targets.items() if v != 0} - results = simulate_execution(non_zero_targets, portfolio_value, config) - # 4) 处理零目标（平仓）：构造 ExecutionResult，shares = held（全部卖出） - for stock_code, weight in effective_targets.items(): - if weight == 0 and stock_code in holdings and holdings[stock_code] > 0: - price = prices.get(stock_code, 0.0) - if price > 0: - held = holdings[stock_code] - # 全部卖出：target_shares = held - # executed_value = held * price（考虑滑点） - slippage_factor = 1.0 - config.slippage_bps / 10000.0 - target_value = -held * price - executed_value = target_value * slippage_factor - commission = abs(executed_value) * config.commission_bps / 10000.0 - stamp_tax = abs(executed_value) * config.stamp_tax_bps / 10000.0 - slippage_cost = abs(executed_value - target_value) - # 标记净卖出 shares = held - results.append( - ExecutionResult( - stock_code=stock_code, - target_value=target_value, - executed_value=executed_value, - commission=commission, - stamp_tax=stamp_tax, - slippage_cost=slippage_cost, - total_cost=commission + stamp_tax + slippage_cost, - net_cash_flow=executed_value - commission - stamp_tax, - ) - ) - # 5) 应用执行结果到持仓 - for r in results: - cost = r.executed_value + r.commission + r.stamp_tax - proceeds = r.executed_value - r.commission - r.stamp_tax - price = prices.get(r.stock_code, 0.0) - if r.target_value > 0: - # 买入：shares = 正数 executed_value / price，cash 减少 cost - shares = r.executed_value / price if price > 0 else 0.0 - holdings[r.stock_code] = holdings.get(r.stock_code, 0.0) + shares - cash -= cost - else: - # 卖出：cash 增加 proceeds 的绝对值（proceeds 本是负的） - held = holdings.get(r.stock_code, 0.0) - if held > 0: - # 如果是 zero-target 触发的全卖（target_value 与持仓市值近似），全部卖出 - if abs(r.target_value) >= held * price * 0.95: - sell_shares = held - else: - target_shares = abs(r.executed_value) / price if price > 0 else held - sell_shares = min(held, target_shares) - holdings[r.stock_code] = held - sell_shares - if holdings[r.stock_code] < 1e-6: - del holdings[r.stock_code] - # proceeds 是负的（target_value 负），cash += proceeds 实际是减去 - # 但卖出是现金流入，所以应该 cash += abs(proceeds) - cash += abs(proceeds) - return positions + + return ExecutionSimulationResult( + initial_cash=initial_cash, + positions=tuple(positions), + daily_executions=tuple(daily_executions), + ) + + +def simulate_multi_day( + target_weights_history: list[tuple[str, dict[str, float]]], + price_history: list[tuple[str, dict[str, float]]], + initial_cash: float, + config: ExecutionConfig | None = None, +) -> list[DailyPosition]: + """兼容入口：返回多日仿真的日末持仓快照。""" + result = simulate_multi_day_with_audit( + target_weights_history, + price_history, + initial_cash, + config, + ) + return list(result.positions) def run_end_to_end_poc( @@ -486,64 +576,16 @@ def run_end_to_end_poc( config = ExecutionConfig() if len(signals) != len(prices): raise ValueError("signals and prices must have same length") - positions = simulate_multi_day(signals, prices, initial_cash, config) - nav_series = pd.Series( - [p.portfolio_value for p in positions], index=[p.date for p in positions] - ) - # 计算 total_costs / total_turnover（重放所有执行） - total_cost_acc = 0.0 - total_turnover_acc = 0.0 - rebalance_count = 0 - cash = initial_cash - holdings: dict[str, float] = {} - for (date, targets), (_, price_map) in zip(signals, prices, strict=True): - portfolio_value = cash + sum( - shares * price_map.get(code, 0.0) for code, shares in holdings.items() - ) - if targets: - rebalance_count += 1 - # 自动平仓：持仓但不在 target 中的股票 - effective_targets: dict[str, float] = dict(targets) - for held_code in holdings: - if held_code not in effective_targets: - effective_targets[held_code] = 0.0 - results = simulate_execution(effective_targets, portfolio_value, config) - total_cost_acc += total_costs(results) - total_turnover_acc += total_turnover(results) - for r in results: - cost = r.executed_value + r.commission + r.stamp_tax - proceeds = r.executed_value - r.commission - r.stamp_tax - if r.target_value > 0: - shares = ( - r.executed_value / price_map[r.stock_code] - if price_map[r.stock_code] > 0 - else 0.0 - ) - holdings[r.stock_code] = holdings.get(r.stock_code, 0.0) + shares - cash -= cost - else: - held = holdings.get(r.stock_code, 0.0) - if held > 0: - sell_shares = min( - held, - abs(r.executed_value / price_map[r.stock_code]) - if price_map[r.stock_code] > 0 - else held, - ) - holdings[r.stock_code] = held - sell_shares - if holdings[r.stock_code] < 1e-6: - del holdings[r.stock_code] - cash += proceeds + audit = simulate_multi_day_with_audit(signals, prices, initial_cash, config) return { - "positions": positions, - "nav_series": nav_series, - "total_costs": total_cost_acc, - "total_turnover": total_turnover_acc, - "total_rebalances": rebalance_count, - "final_portfolio_value": nav_series.iloc[-1] if len(nav_series) > 0 else initial_cash, - "return_pct": ((nav_series.iloc[-1] / initial_cash) - 1) * 100 - if len(nav_series) > 0 - else 0.0, + "positions": list(audit.positions), + "daily_executions": list(audit.daily_executions), + "nav_series": audit.nav_series, + "total_costs": audit.total_costs, + "total_turnover": audit.total_turnover, + "total_rebalances": audit.total_rebalances, + "final_portfolio_value": audit.final_portfolio_value, + "return_pct": audit.return_pct, } @@ -667,7 +709,9 @@ __all__ = [ "apply_bid_ask_spread", "DailyPosition", "DailyExecution", + "ExecutionSimulationResult", "simulate_multi_day", + "simulate_multi_day_with_audit", "run_end_to_end_poc", "DailyPnL", "simulate_with_daily_data", diff --git a/tests/test_execution.py b/tests/test_execution.py index ad26c50..c90eacc 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -329,12 +329,18 @@ def test_simulate_multi_day_length_mismatch_raises(): def test_simulate_multi_day_first_day_value_equals_initial(): - """第一天 portfolio_value = initial_cash（无持仓）。""" + """零成本下第一天日末 NAV 等于初始资金。""" signals = [("d1", {"A": 1.0})] prices = [("d1", {"A": 10.0})] - positions = simulate_multi_day(signals, prices, 1_000_000.0) - # 第一天 NAV = 1_000_000（无持仓），第二天才是调仓后 + config = ExecutionConfig( + commission_bps=0, + stamp_tax_bps=0, + slippage_bps=0, + min_trade_amount=0, + ) + positions = simulate_multi_day(signals, prices, 1_000_000.0, config) assert positions[0].portfolio_value == 1_000_000.0 + assert positions[0].holdings == {"A": 100_000.0} def test_simulate_multi_day_holdings_evolution(): @@ -810,8 +816,8 @@ def test_compute_realized_pnl_sell_realizes(): target_weights_history=targets, ) pnl_list = compute_realized_pnl(positions) - # 第三天（卖出兑现）应有 realized 正利润（cash 从 -800 → 2M = +2M） - assert pnl_list[2].realized_pnl > 0 + # 第二天日末快照已包含当日卖出，现金流入应在当天反映。 + assert pnl_list[1].realized_pnl > 0 # ── O3: end-to-end 端到端测试（集成多个函数） ──────────────