fix: derive multi-day trades from target-weight deltas

2026-08-21 21:35:15 +08:00
parent c51d803daf
commit 17b680604d
2 changed files with 191 additions and 141 deletions
@@ -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",
@@ -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 端到端测试（集成多个函数） ──────────────