Compare commits

...
Author SHA1 Message Date
ao gong 5fb4b85cc3 wip: hand off stacked daily ledger 2026-08-21 22:06:51 +08:00
ao gong a421278527 fix: align ledger performance with signal window 2026-08-21 22:05:28 +08:00
ao gong c774a4546a test: exclude factor warmup from performance window 2026-08-21 22:04:52 +08:00
ao gong 022b87fdac feat: expose platform-neutral ledger projection 2026-08-21 22:03:21 +08:00
ao gong 1b39c53f18 test: define daily ledger projection contract 2026-08-21 22:02:58 +08:00
ao gong b794ab2e8f feat: add post-execution daily ledger 2026-08-21 22:01:35 +08:00
ao gong a44e2d306a test: define post-execution daily ledger contract 2026-08-21 21:58:01 +08:00
ao gong 15b283bdf9 docs: distinguish signal execution and holding times
CI / lite (pull_request) Successful in 4s
2026-08-21 21:49:07 +08:00
ao gong f9b7f2ab1a feat: schedule factor weights for next-session execution 2026-08-21 21:47:53 +08:00
ao gong 213aa88deb test: require execution-price input snapshot 2026-08-21 21:46:47 +08:00
ao gong b7f77e2a6c test: define lagged factor-to-execution contract 2026-08-21 21:46:00 +08:00
ao gong c9fb5978f0 docs: clarify execution audit result contract
CI / lite (pull_request) Successful in 3s
2026-08-21 21:39:12 +08:00
ao gong b2af2a10a6 docs: document auditable execution workflow 2026-08-21 21:38:08 +08:00
ao gong da2ca51ff7 fix: enforce cash-backed long-only rebalancing 2026-08-21 21:37:02 +08:00
ao gong ee22d1eb9d test: reproduce execution cash and long-only violations 2026-08-21 21:35:47 +08:00
ao gong 17b680604d fix: derive multi-day trades from target-weight deltas 2026-08-21 21:35:15 +08:00
ao gong c51d803daf test: reproduce multi-day execution audit gaps 2026-08-21 21:31:52 +08:00
10 changed files with 1538 additions and 182 deletions
+55 -9
View File
@@ -19,11 +19,12 @@
## 模块
- `alpha_factors` — 158 alpha 公式 + 24 基础算子(移植自 qlib alpha158)
- `execution` — 执行仿真(成本/滑点/T+1/涨跌停/部分成交/价差)+ 多日 NAV + PnL 拆解(借鉴 hikyuu 部件化思想)
- `execution` — A 股长仓执行仿真(成本/滑点/现金约束)+ 稀疏调仓/完整交易日 Ledger + 可投影成交与 NAV 审计;T+1、涨跌停、成交量与价差提供独立约束函数
- `indicators` — 50+ 技术指标(MACD / KDJ / 布林 / ATR / ADX / 等)
- `data_adapter` — 桥接 qtdb_pro 长表与新模块(rename / long-wide / 复权 / vwap 代理)
- `backtest` — weight-based 多日仿真(rebalance_table / compute_nav / compare_to_benchmark)
- `portfolio_construction` — 多期因子分数 → Top-K → 等权目标权重表
- `research_pipeline` — 因子日 → 下一真实交易日 → 显式执行价 → 日末估值 → 成本后绩效(防前视编排)
- `metrics` — 绩效(年化收益 / 波动率 / Sharpe / 最大回撤 / Calmar)
- `factor_library` — 通用方法(turnover / winsorize / IC / OLS / jb_test)
- `portfolio_decomp` — 组合分解(risk_parity / mean_variance / 因子归因)
@@ -59,10 +60,13 @@ ruff check src/ tests/ # lint
```python
from quant_engine.alpha_factors import alpha_001, alpha_005, ALPHA158_REGISTRY
from quant_engine.execution import (
ExecutionConfig, simulate_with_daily_data, compute_realized_pnl,
ExecutionConfig, simulate_daily_ledger_with_audit,
simulate_multi_day_with_audit, simulate_with_daily_data,
)
from quant_engine.research_pipeline import (
run_factor_backtest_research, run_factor_execution_research,
)
from quant_engine.backtest import run_weight_backtest
from quant_engine.portfolio_construction import scores_to_weight_table
from quant_engine.indicators import macd, bollinger, kdj
from quant_engine.data_adapter import (
long_to_wide, wide_to_long, rename_tushare_columns,
@@ -73,21 +77,63 @@ from quant_engine.data_adapter import (
# 端到端:qtdb_pro 长表 → 适配 → alpha158 → execution
df = load_qtdb_daily(["000001.SZ"], "2024-01-01", with_adj=True)
prices, volumes = prepare_execution_inputs(df)
result = simulate_with_daily_data(prices, initial_cash=1_000_000.0)
close_prices, volumes = prepare_execution_inputs(df)
open_prices, _ = prepare_execution_inputs(df, price_col="open")
result = simulate_with_daily_data(close_prices, initial_cash=1_000_000.0)
# 多期因子分数 → Top-K 等权组合 → 稳定回测结果
rebalance_weights = scores_to_weight_table(
# 已正确滞后的目标权重 → 现金约束执行 → 唯一来源的成交/拒绝/日末持仓/NAV
execution = simulate_multi_day_with_audit(
target_weights_history=[
("2024-01-02", {"000001.SZ": 1.0}),
("2024-01-03", {"000001.SZ": 1.0}),
],
price_history=[
("2024-01-02", {"000001.SZ": 10.0}),
("2024-01-03", {"000001.SZ": 10.5}),
],
initial_cash=1_000_000.0,
config=ExecutionConfig(),
)
print(execution.nav_series)
print(execution.daily_executions)
# 多期因子分数(必须是 point-in-time 数据)→ Top-K → 下一交易日 open 执行
factor_execution = run_factor_execution_research(
factor_scores,
top_k=20,
gross_exposure=1.0,
execution_prices=open_prices,
execution_price_field="open",
initial_cash=1_000_000.0,
)
# 推荐研究入口:同一交易日历上显式区分 open 成交和 close 估值。
# 因子日保持现金,下一交易日成交后的真实持仓才参与当日收盘收益。
factor_backtest = run_factor_backtest_research(
factor_scores,
top_k=20,
execution_prices=open_prices,
valuation_prices=close_prices,
execution_price_field="open",
valuation_price_field="close",
initial_cash=1_000_000.0,
config=ExecutionConfig(),
)
print(factor_backtest.nav)
print(factor_backtest.returns)
print(factor_backtest.stats())
print(factor_backtest.execution.ledger_frame)
print(factor_backtest.execution.trades_frame)
# run_weight_backtest 是低层算子:只接受收益区间开始前已经生效的持仓权重。
# 不要把 signal-date 的 factor_scores/decision_weights 直接传给它。
backtest = run_weight_backtest(
weights=rebalance_weights,
weights=effective_holding_weights,
stock_returns=daily_returns,
initial_capital=1_000_000.0,
benchmark_nav=benchmark_nav,
)
print(factor_execution.schedule.signal_to_execution)
print(factor_execution.execution.daily_executions)
print(backtest.stats())
print(backtest.benchmark_report())
```
@@ -0,0 +1,33 @@
# Post-execution daily Ledger handoff
## 状态
- 分支:`codex/post-execution-ledger-20260821`
- 基线:`codex/core-contracts-20260821`(PR #2,尚未获用户确认合并)
- 本分支不得直接合并到 `main`;先等待 PR #2 合并,再整理基线并创建独立 PR。
- 无账户、券商、数据库或实盘副作用。
## 已完成
- 新增稀疏调仓、完整交易日估值的 `simulate_daily_ledger_with_audit()`。
- 显式分离 execution price 与 valuation price,支持下一日 open 成交、当日 close 估值。
- 成交记录补齐 `side / quantity / price`,并提供 `trades_frame`。
- 提供平台中立的 `ledger_frame`,不携带 `run_id`,不写数据库。
- 新增 `run_factor_backtest_research()`:PIT 因子、下一交易日执行、日频 NAV、首日成本收益和标准绩效。
- 研究区间从首条有效信号日开始,排除因子预热行情对绩效的稀释。
## 验证
- `pytest -q --cov=src --cov-report=term-missing`:500 passed,total coverage 91%。
- `mypy --strict src/`:14 source files passed。
- 本阶段文件 scoped Ruff:passed。
- 全仓 Ruff:仅既有 governance tests 的 13 个 PT009 基线问题。
- workspace verify/status:通过;仅提示功能分支不是引导基线 `main`。
- 全局 Gitea workflow check:通过,23 个既有警告。
## 继续步骤
1. 获得用户对 PR #2 的明确合并确认并按 L2 流程合并。
2. 将本分支整理到更新后的 `main`,重新运行相同全量验证。
3. 为 Ledger 阶段创建独立 PR,执行唯一一次最终 `ship --ready`,等待用户确认合并。
4. 后续在 `research_results` 增加业务投影适配器,再由 `research_platform` 持久化和展示;核心层继续保持无数据库写入。
+3 -2
View File
@@ -76,8 +76,9 @@ def compute_nav_from_weights(
) -> pd.Series:
"""从调仓表(日期 × 股票权重)+ 个股日收益 → 净值曲线。
假设:在调仓日之间权重不变(**前向填充**)。
调仓日的权重 = `weights.loc[rebalance_date]`。
假设:输入是该收益测量区间开始前已经生效的持仓权重,并在调仓日之间
保持不变(**前向填充**)。本函数不会把信号日自动解释为执行日;因子分数
应先经交易日历调度和实际执行时点处理,避免把同一时点未知的收益计入。
Args:
weights: 调仓日 × 股票代码 的权重 DataFrame(**0~1**,行和 ≤ 1)
+7 -4
View File
@@ -287,6 +287,8 @@ def prepare_execution_inputs(
df: pd.DataFrame,
stock_col: str = "stock_code",
date_col: str = "trade_date",
*,
price_col: str = "close",
) -> tuple[pd.DataFrame, pd.DataFrame]:
"""长表行情 → execution 输入(prices + volumes 宽表)。
@@ -294,10 +296,11 @@ def prepare_execution_inputs(
df: 长表行情(含 close / volume 列,Tushare rename 后)
stock_col: 股票代码列名
date_col: 日期列名
price_col: 执行价字段,默认 close;防前视研究可显式选择下一交易日 open
Returns:
(prices_wide, volumes_wide):
- prices_wide: date × stock_code,值=close
- prices_wide: date × stock_code,值=price_col
- volumes_wide: date × stock_code,值=volume(若无 volume 列则全 1.0)
Examples:
@@ -313,9 +316,9 @@ def prepare_execution_inputs(
"""
if df.empty:
return pd.DataFrame(), pd.DataFrame()
if "close" not in df.columns:
raise ValueError(f"prepare_execution_inputs: 缺 close 列,实际列={list(df.columns)}")
prices = long_to_wide(df, value_col="close", date_col=date_col, stock_col=stock_col)
if price_col not in df.columns:
raise ValueError(f"prepare_execution_inputs: 缺 {price_col} 列,实际列={list(df.columns)}")
prices = long_to_wide(df, value_col=price_col, date_col=date_col, stock_col=stock_col)
if "volume" in df.columns:
volumes = long_to_wide(df, value_col="volume", date_col=date_col, stock_col=stock_col)
else:
+493 -145
View File
@@ -10,7 +10,9 @@
借鉴 hikyuu SG/MM/CN/PG 部件化思想(不引入 hikyuu 框架):
- ExecutionConfig:佣金 + 印花税 + 滑点 + 最小交易额 + 止损/止盈阈值
- simulate_execution():从目标权重 → 实际成交金额(应用成本/滑点)
- simulate_multi_day():多日组合仿真(NAV 序列 + 调仓记录)
- simulate_daily_ledger_with_audit():稀疏调仓 + 完整交易日收盘估值 Ledger
- simulate_multi_day_with_audit():目标权重差额调仓(成交/拒绝/持仓/NAV)
- simulate_multi_day():兼容的多日日末持仓快照入口
- check_stop_loss_take_profit():止损/止盈触发判定
- run_end_to_end_poc():signal → 调仓 → 执行 → NAV 端到端 POC
@@ -19,8 +21,9 @@
from __future__ import annotations
import math
from collections.abc import Mapping
from dataclasses import dataclass
from dataclasses import dataclass, replace
from typing import Any
import pandas as pd
@@ -101,6 +104,9 @@ class ExecutionResult:
net_cash_flow: float # 净现金流(买入为负,卖出为正)
partial_fill_pct: float = 1.0 # 实际成交占目标的比例(1.0 = 全部成交)
blocked_reason: str = "" # 阻塞原因(如涨跌停停牌)
side: str = "" # buy / sell;未成交记录也保留目标方向
quantity: float = 0.0 # 实际成交股数
price: float = 0.0 # 未含滑点的参考执行价
def _apply_costs(
@@ -344,19 +350,458 @@ 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 normalized_nav_series(self) -> pd.Series:
"""返回以初始资金为 1 的净值曲线副本。"""
nav = self.nav_series
if self.initial_cash == 0:
return pd.Series(0.0, index=nav.index, dtype=float)
return nav / self.initial_cash
@property
def daily_returns(self) -> pd.Series:
"""返回逐日收益;首日相对初始资金计算,保留首日交易成本。"""
nav = self.nav_series
if nav.empty:
return nav
returns = nav.pct_change()
returns.iloc[0] = (
nav.iloc[0] / self.initial_cash - 1.0 if self.initial_cash != 0 else 0.0
)
return returns.fillna(0.0)
@property
def trades_frame(self) -> pd.DataFrame:
"""返回可投影到平台成交明细的实际成交表,不包含纯拒绝记录。"""
columns = [
"trade_date",
"ts_code",
"side",
"qty",
"price",
"amount",
"fee",
"slippage",
]
rows = [
{
"trade_date": daily.date,
"ts_code": execution.stock_code,
"side": execution.side,
"qty": execution.quantity,
"price": execution.price,
"amount": execution.executed_value,
"fee": execution.commission + execution.stamp_tax,
"slippage": execution.slippage_cost,
}
for daily in self.daily_executions
for execution in daily.executions
if execution.quantity > 0
]
return pd.DataFrame(rows, columns=columns)
@property
def ledger_frame(self) -> pd.DataFrame:
"""返回稳定的日频 Ledger 投影,不附加运行元数据或写数据库。"""
columns = [
"trade_date",
"portfolio_value",
"nav",
"pnl",
"pnl_pct",
"position_value",
"cash",
"turnover",
]
previous_value = self.initial_cash
rows: list[dict[str, float | str]] = []
daily_returns = self.daily_returns
for index, (position, daily) in enumerate(
zip(self.positions, self.daily_executions, strict=True)
):
daily_turnover = sum(
execution.executed_value
for execution in daily.executions
if execution.quantity > 0
)
turnover_rate = daily_turnover / daily.nav_before if daily.nav_before > 0 else 0.0
rows.append(
{
"trade_date": position.date,
"portfolio_value": position.portfolio_value,
"nav": (
position.portfolio_value / self.initial_cash
if self.initial_cash != 0
else 0.0
),
"pnl": position.portfolio_value - previous_value,
"pnl_pct": float(daily_returns.iloc[index]),
"position_value": position.portfolio_value - position.cash,
"cash": position.cash,
"turnover": turnover_rate,
}
)
previous_value = position.portfolio_value
return pd.DataFrame(rows, columns=columns)
@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,
side="buy" if target_value > 0 else "sell" if target_value < 0 else "",
)
def _validate_target_weights(date: str, targets: Mapping[str, float]) -> dict[str, float]:
"""校验并复制单日长仓目标权重。"""
normalized: dict[str, float] = {}
for stock_code, raw_weight in targets.items():
try:
weight = float(raw_weight)
except (TypeError, ValueError) as error:
raise ValueError(f"target weights on {date!r} must be numeric") from error
if not math.isfinite(weight) or weight < 0:
raise ValueError(f"target weights on {date!r} must be finite and non-negative")
normalized[stock_code] = weight
if sum(normalized.values()) > 1.0 + 1e-12:
raise ValueError(f"target weights on {date!r} must sum to at most 1.0")
return normalized
def _partially_fill_buy(
desired: ExecutionResult,
fill_pct: float,
config: ExecutionConfig,
) -> ExecutionResult:
"""按同一比例缩放买入,保留原始目标金额供审计。"""
actual_target_value = desired.target_value * fill_pct
executed_value, commission, stamp_tax, slippage_cost = _apply_costs(
actual_target_value,
True,
config,
)
total_cost = commission + stamp_tax + slippage_cost
return ExecutionResult(
stock_code=desired.stock_code,
target_value=desired.target_value,
executed_value=executed_value,
commission=commission,
stamp_tax=stamp_tax,
slippage_cost=slippage_cost,
total_cost=total_cost,
net_cash_flow=-(executed_value + commission + stamp_tax),
partial_fill_pct=fill_pct,
blocked_reason="insufficient_cash_partial_fill",
)
def _rebalance_at_prices(
date: str,
targets: Mapping[str, float],
prices: Mapping[str, float],
cash: float,
holdings: dict[str, float],
config: ExecutionConfig,
) -> tuple[float, tuple[ExecutionResult, ...], float, float]:
"""在单一执行时点按目标权重差额调仓,并原地更新 holdings。"""
normalized_targets = _validate_target_weights(date, targets)
for held_code in holdings:
held_price = prices.get(held_code)
if held_price is None or not math.isfinite(held_price) or held_price <= 0:
raise ValueError(f"missing price for held asset {held_code} on {date!r}")
nav_before = cash + sum(
shares * prices.get(stock_code, 0.0)
for stock_code, shares in holdings.items()
)
effective_targets = dict.fromkeys(holdings, 0.0)
effective_targets.update(normalized_targets)
buy_weights: dict[str, float] = {}
sell_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
destination = buy_weights if trade_value > 0 else sell_weights
destination[stock_code] = trade_value / nav_before
sell_executions = simulate_execution(sell_weights, nav_before, config)
filled: list[ExecutionResult] = []
for raw_execution in sell_executions:
price = prices[raw_execution.stock_code]
quantity = abs(raw_execution.target_value) / price
execution = replace(
raw_execution,
side="sell",
quantity=quantity,
price=price,
)
held = holdings.get(execution.stock_code, 0.0)
holdings[execution.stock_code] = max(0.0, held - quantity)
if holdings[execution.stock_code] < 1e-6:
del holdings[execution.stock_code]
cash += execution.net_cash_flow
filled.append(execution)
desired_buys = simulate_execution(buy_weights, nav_before, config)
required_cash = sum(-execution.net_cash_flow for execution in desired_buys)
buy_fill_pct = min(1.0, max(cash, 0.0) / required_cash) if required_cash > 0 else 1.0
for desired in desired_buys:
if buy_fill_pct == 0:
rejected.append(
_blocked_execution(desired.stock_code, desired.target_value, "insufficient_cash")
)
continue
raw_execution = (
desired
if buy_fill_pct == 1.0
else _partially_fill_buy(desired, buy_fill_pct, config)
)
price = prices[raw_execution.stock_code]
quantity = abs(raw_execution.target_value) * raw_execution.partial_fill_pct / price
execution = replace(
raw_execution,
side="buy",
quantity=quantity,
price=price,
)
holdings[execution.stock_code] = holdings.get(execution.stock_code, 0.0) + quantity
cash += execution.net_cash_flow
if math.isclose(cash, 0.0, abs_tol=1e-9):
cash = 0.0
filled.append(execution)
executions = (*filled, *rejected)
nav_after = cash + sum(
shares * prices.get(stock_code, 0.0)
for stock_code, shares in holdings.items()
)
return cash, executions, nav_before, nav_after
def _validate_sparse_daily_histories(
target_weights_history: list[tuple[str, dict[str, float]]],
execution_price_history: list[tuple[str, dict[str, float]]],
valuation_price_history: list[tuple[str, dict[str, float]]],
) -> tuple[
dict[str, dict[str, float]],
dict[str, dict[str, float]],
list[tuple[str, dict[str, float]]],
]:
"""校验稀疏调仓与完整估值日历,并隔离调用方可变输入。"""
target_dates = [date for date, _ in target_weights_history]
execution_dates = [date for date, _ in execution_price_history]
valuation_dates = [date for date, _ in valuation_price_history]
if len(set(target_dates)) != len(target_dates):
raise ValueError("target_weights_history must contain unique dates")
if len(set(execution_dates)) != len(execution_dates):
raise ValueError("execution_price_history must contain unique dates")
if len(set(valuation_dates)) != len(valuation_dates):
raise ValueError("valuation_price_history must contain unique dates")
if execution_dates != target_dates:
raise ValueError("execution price dates must exactly match target weight dates")
valuation_positions = {date: index for index, date in enumerate(valuation_dates)}
missing_dates = [date for date in target_dates if date not in valuation_positions]
if missing_dates:
raise ValueError(f"target dates must belong to valuation calendar: {missing_dates}")
positions = [valuation_positions[date] for date in target_dates]
if positions != sorted(positions):
raise ValueError("target weights must follow valuation calendar order")
targets = {date: dict(values) for date, values in target_weights_history}
execution_prices = {date: dict(values) for date, values in execution_price_history}
valuation_prices = [(date, dict(values)) for date, values in valuation_price_history]
return targets, execution_prices, valuation_prices
def _simulate_daily_ledger(
target_weights_history: list[tuple[str, dict[str, float]]],
execution_price_history: list[tuple[str, dict[str, float]]],
valuation_price_history: list[tuple[str, dict[str, float]]],
initial_cash: float,
config: ExecutionConfig,
) -> ExecutionSimulationResult:
targets_by_date, execution_prices_by_date, valuation_history = (
_validate_sparse_daily_histories(
target_weights_history,
execution_price_history,
valuation_price_history,
)
)
cash = initial_cash
holdings: dict[str, float] = {}
positions: list[DailyPosition] = []
daily_executions: list[DailyExecution] = []
for date, valuation_prices in valuation_history:
targets = targets_by_date.get(date)
if targets is None:
executions: tuple[ExecutionResult, ...] = ()
nav_before = 0.0
nav_after = 0.0
rebalance_triggered = False
else:
cash, executions, nav_before, nav_after = _rebalance_at_prices(
date,
targets,
execution_prices_by_date[date],
cash,
holdings,
config,
)
rebalance_triggered = any(execution.quantity > 0 for execution in executions)
for held_code in holdings:
valuation_price = valuation_prices.get(held_code)
if (
valuation_price is None
or not math.isfinite(valuation_price)
or valuation_price <= 0
):
raise ValueError(
f"missing valuation price for held asset {held_code} on {date!r}"
)
portfolio_value = cash + sum(
shares * valuation_prices[stock_code]
for stock_code, shares in holdings.items()
)
if targets is None:
nav_before = portfolio_value
nav_after = portfolio_value
positions.append(DailyPosition(date, cash, dict(holdings), portfolio_value))
daily_executions.append(
DailyExecution(
date=date,
executions=executions,
nav_before=nav_before,
nav_after=nav_after,
rebalance_triggered=rebalance_triggered,
)
)
return ExecutionSimulationResult(
initial_cash=initial_cash,
positions=tuple(positions),
daily_executions=tuple(daily_executions),
)
def simulate_daily_ledger_with_audit(
target_weights_history: list[tuple[str, dict[str, float]]],
execution_price_history: list[tuple[str, dict[str, float]]],
valuation_price_history: list[tuple[str, dict[str, float]]],
initial_cash: float,
config: ExecutionConfig | None = None,
) -> ExecutionSimulationResult:
"""以稀疏调仓和完整日历运行成交后持仓 Ledger。
执行价只用于调仓日现金与股数变化,估值价用于每个交易日日末 NAV;二者
显式分离,从而支持“下一日 open 成交、同日 close 估值”的无前视研究。
"""
if not math.isfinite(initial_cash) or initial_cash <= 0:
raise ValueError(f"initial_cash must be positive and finite, got {initial_cash}")
return _simulate_daily_ledger(
target_weights_history,
execution_price_history,
valuation_price_history,
initial_cash,
ExecutionConfig() if config is None else config,
)
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 +810,45 @@ 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 []
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()
for (date, _), (price_date, _) 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}"
)
positions.append(
DailyPosition(
date=date,
cash=cash,
holdings=dict(holdings),
portfolio_value=portfolio_value,
return _simulate_daily_ledger(
target_weights_history,
price_history,
price_history,
initial_cash,
ExecutionConfig() if config is None else config,
)
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,
)
# 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 list(result.positions)
def run_end_to_end_poc(
@@ -486,64 +879,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 +1012,10 @@ __all__ = [
"apply_bid_ask_spread",
"DailyPosition",
"DailyExecution",
"ExecutionSimulationResult",
"simulate_daily_ledger_with_audit",
"simulate_multi_day",
"simulate_multi_day_with_audit",
"run_end_to_end_poc",
"DailyPnL",
"simulate_with_daily_data",
+347
View File
@@ -0,0 +1,347 @@
"""可信研究链路:因子分数经交易日历滞后后进入执行与日频 Ledger。
本模块只编排现有组合构建与执行组件,不连接账户、券商或实盘订单。
时间契约借鉴 Qlib 的 prediction/trade time 分离与 Backtrader 的 next-bar
执行语义:signal_date 上形成的目标权重,默认最早在下一交易时点执行。
完整回测链路进一步分离 execution price 与日末 valuation price,非调仓日也
持续盯市,并从真实成交后持仓派生日收益和绩效。
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
import numpy as np
import pandas as pd
from pandas.api.types import is_numeric_dtype
from quant_engine.execution import (
ExecutionConfig,
ExecutionSimulationResult,
simulate_daily_ledger_with_audit,
simulate_multi_day_with_audit,
)
from quant_engine.metrics import summary as metrics_summary
from quant_engine.portfolio_construction import scores_to_weight_table
__all__ = [
"TargetWeightSchedule",
"FactorExecutionResult",
"FactorBacktestResult",
"schedule_target_weights",
"run_factor_execution_research",
"run_factor_backtest_research",
]
@dataclass(frozen=True, slots=True, eq=False)
class TargetWeightSchedule:
"""保留决策时间和执行时间的目标权重调度快照。"""
decision_weights: pd.DataFrame
signal_to_execution: pd.Series
execution_weights: pd.DataFrame
lag_sessions: int
@dataclass(frozen=True, slots=True, eq=False)
class FactorExecutionResult:
"""因子到执行审计的一次可复现研究结果。"""
factor_scores: pd.DataFrame
execution_prices: pd.DataFrame
schedule: TargetWeightSchedule
execution_price_field: str
execution: ExecutionSimulationResult
@dataclass(frozen=True, slots=True, eq=False)
class FactorBacktestResult:
"""因子、成交后日频 Ledger 与绩效的一次可复现快照。"""
factor_scores: pd.DataFrame
execution_prices: pd.DataFrame
valuation_prices: pd.DataFrame
schedule: TargetWeightSchedule
execution_price_field: str
valuation_price_field: str
execution: ExecutionSimulationResult
@property
def nav(self) -> pd.Series:
"""返回以初始资金归一化为 1 的日频 NAV。"""
return pd.Series(
self.execution.normalized_nav_series.to_numpy(copy=True),
index=self.valuation_prices.index.copy(),
name="nav",
)
@property
def returns(self) -> pd.Series:
"""返回包含首日成本影响的日频收益。"""
return pd.Series(
self.execution.daily_returns.to_numpy(copy=True),
index=self.valuation_prices.index.copy(),
name="returns",
)
def stats(self, rf: float = 0.0) -> Mapping[str, float]:
"""复用标准绩效口径计算指标。"""
return metrics_summary(self.returns, rf)
def _validate_datetime_index(index: pd.Index, name: str) -> pd.DatetimeIndex:
if not isinstance(index, pd.DatetimeIndex):
raise TypeError(f"{name} must use a DatetimeIndex")
if not index.is_unique:
raise ValueError(f"{name} must contain unique sessions")
if not index.is_monotonic_increasing:
raise ValueError(f"{name} must be in chronological order")
return index
def _validate_decision_weights(decision_weights: pd.DataFrame) -> None:
if not isinstance(decision_weights, pd.DataFrame):
raise TypeError(
f"decision_weights must be a pandas DataFrame, got {type(decision_weights).__name__}"
)
_validate_datetime_index(decision_weights.index, "decision_weights index")
if not decision_weights.columns.is_unique:
raise ValueError("decision_weights must contain unique asset labels")
if not all(is_numeric_dtype(dtype) for dtype in decision_weights.dtypes):
raise TypeError("decision_weights must contain numeric values")
values = decision_weights.to_numpy(dtype=float)
if not np.isfinite(values).all() or (values < 0).any():
raise ValueError("decision_weights must be finite and non-negative")
if (decision_weights.sum(axis=1) > 1.0 + 1e-12).any():
raise ValueError("decision_weights rows must sum to at most 1.0")
def _validate_execution_prices(execution_prices: pd.DataFrame) -> pd.DatetimeIndex:
if not isinstance(execution_prices, pd.DataFrame):
raise TypeError(
f"execution_prices must be a pandas DataFrame, got {type(execution_prices).__name__}"
)
calendar = _validate_datetime_index(execution_prices.index, "execution_prices index")
if not execution_prices.columns.is_unique:
raise ValueError("execution_prices must contain unique asset labels")
if not all(is_numeric_dtype(dtype) for dtype in execution_prices.dtypes):
raise TypeError("execution_prices must contain numeric values")
return calendar
def schedule_target_weights(
decision_weights: pd.DataFrame,
trading_calendar: pd.DatetimeIndex,
*,
lag_sessions: int = 1,
) -> TargetWeightSchedule:
"""将信号日目标权重映射到后续真实交易日,不做整数行盲移位。
所有信号日必须属于 ``trading_calendar``,且日历必须包含每个信号对应的
未来执行日;无法执行的末尾信号会显式失败,避免被静默丢弃。
"""
_validate_decision_weights(decision_weights)
calendar = _validate_datetime_index(trading_calendar, "trading_calendar")
if isinstance(lag_sessions, bool) or not isinstance(lag_sessions, int) or lag_sessions <= 0:
raise ValueError("lag_sessions must be a positive integer")
decision_snapshot = decision_weights.copy(deep=True)
if decision_snapshot.empty:
execution_weights = decision_snapshot.copy(deep=True)
execution_weights.index = pd.DatetimeIndex([], name="execution_date")
mapping = pd.Series(
calendar[:0],
index=decision_snapshot.index.copy(),
name="execution_date",
)
return TargetWeightSchedule(
decision_weights=decision_snapshot,
signal_to_execution=mapping,
execution_weights=execution_weights,
lag_sessions=lag_sessions,
)
signal_positions = calendar.get_indexer(decision_snapshot.index)
if (signal_positions < 0).any():
missing = decision_snapshot.index[signal_positions < 0]
raise ValueError(
"signal dates must be trading sessions; missing="
+ ", ".join(str(date) for date in missing)
)
execution_positions = signal_positions + lag_sessions
if (execution_positions >= len(calendar)).any():
unavailable = decision_snapshot.index[execution_positions >= len(calendar)]
raise ValueError(
"trading_calendar lacks a future execution session for signal dates: "
+ ", ".join(str(date) for date in unavailable)
)
execution_dates = calendar.take(execution_positions)
signal_to_execution = pd.Series(
execution_dates,
index=decision_snapshot.index.copy(),
name="execution_date",
)
execution_weights = decision_snapshot.copy(deep=True)
execution_weights.index = pd.DatetimeIndex(execution_dates, name="execution_date")
return TargetWeightSchedule(
decision_weights=decision_snapshot,
signal_to_execution=signal_to_execution,
execution_weights=execution_weights,
lag_sessions=lag_sessions,
)
def run_factor_execution_research(
factor_scores: pd.DataFrame,
execution_prices: pd.DataFrame,
*,
top_k: int,
execution_price_field: str,
lag_sessions: int = 1,
gross_exposure: float = 1.0,
largest: bool = True,
initial_cash: float = 1_000_000.0,
config: ExecutionConfig | None = None,
) -> FactorExecutionResult:
"""运行因子分数 → 目标权重 → 下一交易时点 → 执行审计链路。
``execution_prices`` 必须代表实际拟执行时点的价格矩阵,例如日频研究中
signal 日收盘生成分数后使用下一交易日 ``open``。价格字段名称被保存在
结果元数据中,但函数不会猜测或重写价格语义。
"""
price_field = execution_price_field.strip()
if not price_field:
raise ValueError("execution_price_field must be non-empty")
calendar = _validate_execution_prices(execution_prices)
factor_snapshot = factor_scores.copy(deep=True)
decision_weights = scores_to_weight_table(
factor_snapshot,
top_k,
gross_exposure=gross_exposure,
largest=largest,
)
schedule = schedule_target_weights(
decision_weights,
calendar,
lag_sessions=lag_sessions,
)
price_snapshot = execution_prices.copy(deep=True)
target_history: list[tuple[str, dict[str, float]]] = []
price_history: list[tuple[str, dict[str, float]]] = []
for execution_date, weights in schedule.execution_weights.iterrows():
date_label = str(pd.Timestamp(execution_date))
target_history.append(
(date_label, {asset: float(weight) for asset, weight in weights.items()})
)
prices = price_snapshot.loc[execution_date]
price_history.append(
(date_label, {asset: float(price) for asset, price in prices.items()})
)
execution = simulate_multi_day_with_audit(
target_history,
price_history,
initial_cash,
config,
)
return FactorExecutionResult(
factor_scores=factor_snapshot,
execution_prices=price_snapshot,
schedule=schedule,
execution_price_field=price_field,
execution=execution,
)
def run_factor_backtest_research(
factor_scores: pd.DataFrame,
execution_prices: pd.DataFrame,
valuation_prices: pd.DataFrame,
*,
top_k: int,
execution_price_field: str,
valuation_price_field: str,
lag_sessions: int = 1,
gross_exposure: float = 1.0,
largest: bool = True,
initial_cash: float = 1_000_000.0,
config: ExecutionConfig | None = None,
) -> FactorBacktestResult:
"""运行 PIT 因子到成交后日频 Ledger、收益与绩效的可信研究链路。"""
execution_field = execution_price_field.strip()
valuation_field = valuation_price_field.strip()
if not execution_field:
raise ValueError("execution_price_field must be non-empty")
if not valuation_field:
raise ValueError("valuation_price_field must be non-empty")
execution_calendar = _validate_execution_prices(execution_prices)
valuation_calendar = _validate_execution_prices(valuation_prices)
if not execution_calendar.equals(valuation_calendar):
raise ValueError("execution and valuation prices must use matching trading calendars")
if not execution_prices.columns.equals(valuation_prices.columns):
raise ValueError("execution and valuation prices must use matching asset labels")
factor_snapshot = factor_scores.copy(deep=True)
execution_snapshot = execution_prices.copy(deep=True)
valuation_snapshot = valuation_prices.copy(deep=True)
decision_weights = scores_to_weight_table(
factor_snapshot,
top_k,
gross_exposure=gross_exposure,
largest=largest,
)
schedule = schedule_target_weights(
decision_weights,
execution_calendar,
lag_sessions=lag_sessions,
)
if decision_weights.empty:
execution_window = execution_snapshot.iloc[:0].copy()
valuation_window = valuation_snapshot.iloc[:0].copy()
else:
research_start = decision_weights.index[0]
execution_window = execution_snapshot.loc[research_start:].copy()
valuation_window = valuation_snapshot.loc[research_start:].copy()
target_history: list[tuple[str, dict[str, float]]] = []
execution_history: list[tuple[str, dict[str, float]]] = []
for execution_date, weights in schedule.execution_weights.iterrows():
date_label = str(pd.Timestamp(execution_date))
target_history.append(
(date_label, {asset: float(weight) for asset, weight in weights.items()})
)
prices = execution_window.loc[execution_date]
execution_history.append(
(date_label, {asset: float(price) for asset, price in prices.items()})
)
valuation_history = [
(
str(pd.Timestamp(valuation_date)),
{asset: float(price) for asset, price in prices.items()},
)
for valuation_date, prices in valuation_window.iterrows()
]
execution = simulate_daily_ledger_with_audit(
target_history,
execution_history,
valuation_history,
initial_cash,
config,
)
return FactorBacktestResult(
factor_scores=factor_snapshot,
execution_prices=execution_window,
valuation_prices=valuation_window,
schedule=schedule,
execution_price_field=execution_field,
valuation_price_field=valuation_field,
execution=execution,
)
+19
View File
@@ -266,6 +266,17 @@ def test_prepare_execution_inputs_basic(tushare_long: pd.DataFrame) -> None:
assert volumes.iloc[0, 0] == pytest.approx(1000.0)
def test_prepare_execution_inputs_can_select_next_session_open_price(
tushare_long: pd.DataFrame,
) -> None:
"""显式 price_col=open 时应生成开盘执行价矩阵。"""
renamed = rename_tushare_columns(tushare_long)
prices, _volumes = prepare_execution_inputs(renamed, price_col="open")
assert prices.iloc[0, 0] == pytest.approx(10.0)
def test_prepare_execution_inputs_no_volume() -> None:
"""无 volume 列 → volumes 全 1.0。"""
df = pd.DataFrame(
@@ -286,6 +297,14 @@ def test_prepare_execution_inputs_missing_close_raises() -> None:
prepare_execution_inputs(df)
def test_prepare_execution_inputs_missing_selected_price_raises() -> None:
df = pd.DataFrame(
{"stock_code": ["A"], "trade_date": ["2024-01-01"], "close": [10.0]}
)
with pytest.raises(ValueError, match="缺 open"):
prepare_execution_inputs(df, price_col="open")
# ── 端到端:长表 → 适配 → alpha158 + execution ──────────────
+307 -14
View File
@@ -11,6 +11,7 @@ import pytest
from quant_engine.execution import (
ExecutionConfig,
ExecutionResult,
ExecutionSimulationResult,
apply_bid_ask_spread,
apply_volume_constraint,
check_price_limit,
@@ -19,7 +20,9 @@ from quant_engine.execution import (
compute_realized_pnl,
run_end_to_end_poc,
simulate_execution,
simulate_daily_ledger_with_audit,
simulate_multi_day,
simulate_multi_day_with_audit,
simulate_with_daily_data,
total_costs,
total_turnover,
@@ -327,24 +330,26 @@ 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():
"""调仓后 holdings 演化。
注意:positions[i] 是第 i 天 rebalance 之前的快照。
所以要看 d2 rebalance 后的 holdings,需要看 positions[2](d3 的快照)。
"""
"""日末快照应反映当天调仓后的 holdings。"""
signals = [
("d1", {"A": 0.5, "B": 0.5}),
("d2", {"A": 1.0, "B": 0.0}), # 全仓 A
("d3", {"A": 1.0, "B": 0.0}), # 第三天的快照才能看到 d2 rebalance 后的 holdings
("d3", {"A": 1.0, "B": 0.0}),
]
prices = [
("d1", {"A": 10.0, "B": 20.0}),
@@ -352,9 +357,288 @@ def test_simulate_multi_day_holdings_evolution():
("d3", {"A": 12.0, "B": 22.0}),
]
positions = simulate_multi_day(signals, prices, 1_000_000.0)
# d3 的 PRE-trade snapshot 应该只有 A(B 在 d2 被平仓)
assert "B" not in positions[2].holdings
assert "A" in positions[2].holdings
assert "B" not in positions[1].holdings
assert "A" in positions[1].holdings
def test_simulate_multi_day_with_audit_rebalances_target_weights_by_delta():
"""相同目标权重不应在每个交易日重复买入。"""
config = ExecutionConfig(
commission_bps=0,
stamp_tax_bps=0,
slippage_bps=0,
min_trade_amount=0,
)
targets = [(date, {"A": 1.0}) for date in ("d1", "d2", "d3")]
prices = [(date, {"A": 10.0}) for date in ("d1", "d2", "d3")]
result = simulate_multi_day_with_audit(targets, prices, 1_000.0, config)
assert isinstance(result, ExecutionSimulationResult)
assert [len(day.executions) for day in result.daily_executions] == [1, 0, 0]
assert result.total_turnover == pytest.approx(1_000.0)
assert [position.cash for position in result.positions] == pytest.approx([0.0, 0.0, 0.0])
assert [position.holdings["A"] for position in result.positions] == pytest.approx(
[100.0, 100.0, 100.0]
)
assert [position.portfolio_value for position in result.positions] == pytest.approx(
[1_000.0, 1_000.0, 1_000.0]
)
def test_simulate_multi_day_with_audit_records_costs_without_replay():
"""成交成本与日末 NAV 应来自同一次状态推进。"""
targets = [("d1", {"A": 1.0}), ("d2", {"A": 1.0})]
prices = [("d1", {"A": 10.0}), ("d2", {"A": 10.0})]
result = simulate_multi_day_with_audit(targets, prices, 1_000.0)
first_day = result.daily_executions[0]
assert first_day.nav_before == pytest.approx(1_000.0)
assert first_day.nav_after == pytest.approx(result.positions[0].portfolio_value)
assert result.total_costs == pytest.approx(sum(r.total_cost for r in first_day.executions))
assert result.final_portfolio_value == pytest.approx(1_000.0 - result.total_costs)
assert result.daily_executions[1].executions == ()
def test_simulate_multi_day_with_audit_never_spends_more_cash_than_available():
"""满仓目标应按可用现金部分成交,不能用负现金隐式加杠杆。"""
result = simulate_multi_day_with_audit(
[("d1", {"A": 1.0})],
[("d1", {"A": 10.0})],
1_000.0,
)
execution = result.daily_executions[0].executions[0]
assert result.positions[0].cash >= -1e-9
assert 0 < execution.partial_fill_pct < 1
assert execution.blocked_reason == "insufficient_cash_partial_fill"
assert result.final_portfolio_value == pytest.approx(1_000.0 - result.total_costs)
@pytest.mark.parametrize(
"targets",
[
{"A": -0.1},
{"A": 0.6, "B": 0.5},
{"A": float("nan")},
],
)
def test_simulate_multi_day_with_audit_rejects_invalid_long_only_weights(targets):
"""多日 A 股目标必须是有限、非负且合计不超过 100% 的权重。"""
with pytest.raises(ValueError, match="target weights"):
simulate_multi_day_with_audit(
[("d1", targets)],
[("d1", {"A": 10.0, "B": 10.0})],
1_000.0,
)
def test_simulate_multi_day_with_audit_requires_price_for_existing_holding():
"""已有持仓缺价时无法可信估值,必须失败而不是把市值记为零。"""
with pytest.raises(ValueError, match="missing price for held asset A"):
simulate_multi_day_with_audit(
[("d1", {"A": 1.0}), ("d2", {"A": 1.0})],
[("d1", {"A": 10.0}), ("d2", {})],
1_000.0,
)
def test_simulate_multi_day_with_audit_records_unpriced_target_rejection():
"""缺失价格的目标不能吞掉现金,且必须留下拒绝原因。"""
result = simulate_multi_day_with_audit(
[("d1", {"A": 1.0})],
[("d1", {"B": 10.0})],
1_000.0,
)
rejection = result.daily_executions[0].executions[0]
assert rejection.stock_code == "A"
assert rejection.executed_value == 0.0
assert rejection.partial_fill_pct == 0.0
assert rejection.blocked_reason == "missing_price"
assert result.positions[0].cash == 1_000.0
assert result.positions[0].holdings == {}
def test_simulate_multi_day_with_audit_requires_matching_dates():
"""权重与价格日期错位必须显式失败,不能按位置静默配对。"""
with pytest.raises(ValueError, match="dates must match"):
simulate_multi_day_with_audit(
[("d1", {"A": 1.0})],
[("d2", {"A": 10.0})],
1_000.0,
)
# ── 逐交易日 Ledger:成交时点与估值时点分离 ─────────────────
def test_daily_ledger_marks_every_session_after_sparse_open_execution() -> None:
"""下一日开盘成交后,应按每日收盘价持续盯市,而非只记录调仓日。"""
config = ExecutionConfig(
commission_bps=0,
stamp_tax_bps=0,
slippage_bps=0,
min_trade_amount=0,
)
result = simulate_daily_ledger_with_audit(
target_weights_history=[("d1", {"A": 1.0})],
execution_price_history=[("d1", {"A": 10.0})],
valuation_price_history=[
("d0", {"A": 9.0}),
("d1", {"A": 11.0}),
("d2", {"A": 12.0}),
],
initial_cash=1_000.0,
config=config,
)
assert [position.date for position in result.positions] == ["d0", "d1", "d2"]
assert [position.portfolio_value for position in result.positions] == pytest.approx(
[1_000.0, 1_100.0, 1_200.0]
)
assert [len(day.executions) for day in result.daily_executions] == [0, 1, 0]
fill = result.daily_executions[1].executions[0]
assert fill.side == "buy"
assert fill.quantity == pytest.approx(100.0)
assert fill.price == pytest.approx(10.0)
pd.testing.assert_series_equal(
result.normalized_nav_series,
pd.Series([1.0, 1.1, 1.2], index=["d0", "d1", "d2"], dtype=float),
)
pd.testing.assert_series_equal(
result.daily_returns,
pd.Series([0.0, 0.1, 1.2 / 1.1 - 1.0], index=["d0", "d1", "d2"]),
)
def test_daily_ledger_first_session_cost_reduces_first_return() -> None:
"""首个估值日发生交易时,费用必须进入相对初始资金的首日收益。"""
result = simulate_daily_ledger_with_audit(
target_weights_history=[("d0", {"A": 1.0})],
execution_price_history=[("d0", {"A": 10.0})],
valuation_price_history=[("d0", {"A": 10.0})],
initial_cash=1_000.0,
)
assert result.total_costs > 0
assert result.daily_returns.iloc[0] == pytest.approx(
result.final_portfolio_value / result.initial_cash - 1.0
)
assert result.daily_returns.iloc[0] < 0
def test_daily_ledger_nav_is_rebuildable_and_trades_are_projectable() -> None:
"""Ledger 必须同时支持现金守恒校验和平台成交表投影。"""
config = ExecutionConfig(
commission_bps=0,
stamp_tax_bps=0,
slippage_bps=0,
min_trade_amount=0,
)
result = simulate_daily_ledger_with_audit(
target_weights_history=[
("d1", {"A": 1.0, "B": 0.0}),
("d2", {"A": 0.0, "B": 1.0}),
],
execution_price_history=[
("d1", {"A": 10.0, "B": 20.0}),
("d2", {"A": 11.0, "B": 22.0}),
],
valuation_price_history=[
("d0", {"A": 9.0, "B": 19.0}),
("d1", {"A": 10.5, "B": 21.0}),
("d2", {"A": 12.0, "B": 24.0}),
],
initial_cash=1_000.0,
config=config,
)
close_prices = {
"d0": {"A": 9.0, "B": 19.0},
"d1": {"A": 10.5, "B": 21.0},
"d2": {"A": 12.0, "B": 24.0},
}
for position in result.positions:
rebuilt = position.cash + sum(
shares * close_prices[position.date][asset]
for asset, shares in position.holdings.items()
)
assert position.portfolio_value == pytest.approx(rebuilt)
trades = result.trades_frame
assert trades.columns.tolist() == [
"trade_date",
"ts_code",
"side",
"qty",
"price",
"amount",
"fee",
"slippage",
]
assert trades["side"].tolist() == ["buy", "sell", "buy"]
assert (trades["qty"] > 0).all()
def test_daily_ledger_frame_matches_platform_projection_contract() -> None:
"""核心层输出稳定日频投影,但不携带 run_id 或执行数据库写入。"""
config = ExecutionConfig(
commission_bps=0,
stamp_tax_bps=0,
slippage_bps=0,
min_trade_amount=0,
)
result = simulate_daily_ledger_with_audit(
target_weights_history=[("d1", {"A": 1.0})],
execution_price_history=[("d1", {"A": 10.0})],
valuation_price_history=[
("d0", {"A": 9.0}),
("d1", {"A": 11.0}),
("d2", {"A": 12.0}),
],
initial_cash=1_000.0,
config=config,
)
ledger = result.ledger_frame
assert ledger.columns.tolist() == [
"trade_date",
"portfolio_value",
"nav",
"pnl",
"pnl_pct",
"position_value",
"cash",
"turnover",
]
assert ledger["trade_date"].tolist() == ["d0", "d1", "d2"]
assert ledger["nav"].tolist() == pytest.approx([1.0, 1.1, 1.2])
assert ledger["pnl"].tolist() == pytest.approx([0.0, 100.0, 100.0])
assert ledger["pnl_pct"].tolist() == pytest.approx([0.0, 0.1, 1.2 / 1.1 - 1.0])
assert ledger["position_value"].tolist() == pytest.approx([0.0, 1_100.0, 1_200.0])
assert ledger["cash"].tolist() == pytest.approx([1_000.0, 0.0, 0.0])
assert ledger["turnover"].tolist() == pytest.approx([0.0, 1.0, 0.0])
def test_daily_ledger_rejects_missing_close_for_held_asset() -> None:
"""已有持仓缺少收盘估值价时必须 fail closed。"""
with pytest.raises(ValueError, match="missing valuation price for held asset A"):
simulate_daily_ledger_with_audit(
target_weights_history=[("d0", {"A": 1.0})],
execution_price_history=[("d0", {"A": 10.0})],
valuation_price_history=[("d0", {"A": 10.0}), ("d1", {})],
initial_cash=1_000.0,
)
def test_daily_ledger_requires_positive_initial_cash() -> None:
"""可信收益曲线需要正初始资金作为归一化基准。"""
with pytest.raises(ValueError, match="initial_cash must be positive"):
simulate_daily_ledger_with_audit([], [], [], initial_cash=0.0)
# ── v1.2.0 Phase 1:端到端 POC(run_end_to_end_poc) ─────
@@ -449,6 +733,15 @@ def test_run_end_to_end_poc_costs_recorded():
result = run_end_to_end_poc(signals, prices, 1_000_000.0)
assert result["total_costs"] > 0
assert result["total_turnover"] > 0
executions = [
execution
for daily in result["daily_executions"]
for execution in daily.executions
]
assert result["total_costs"] == pytest.approx(sum(item.total_cost for item in executions))
assert result["total_turnover"] == pytest.approx(
sum(item.executed_value for item in executions)
)
# ── v1.2.0 Phase 2: T+1 / 涨跌停 / 部分成交 / 买卖价差 ─────
@@ -741,8 +1034,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 端到端测试(集成多个函数) ──────────────
+5 -6
View File
@@ -91,10 +91,10 @@ def test_scores_to_weight_table_constructs_each_rebalance_independently() -> Non
pd.testing.assert_series_equal(result.iloc[0], changed_result.iloc[0])
def test_factor_scores_flow_directly_into_weight_backtest() -> None:
def test_effective_holding_weights_flow_into_weight_backtest() -> None:
dates = pd.date_range("2026-01-05", periods=3, freq="B")
scores = pd.DataFrame(
{"A": [2.0, 0.0], "B": [1.0, 3.0]},
effective_weights = pd.DataFrame(
{"A": [1.0, 0.0], "B": [0.0, 1.0]},
index=dates[[0, 2]],
)
stock_returns = pd.DataFrame(
@@ -102,11 +102,10 @@ def test_factor_scores_flow_directly_into_weight_backtest() -> None:
index=dates,
)
weights = scores_to_weight_table(scores, top_k=1)
result = run_weight_backtest(weights, stock_returns)
result = run_weight_backtest(effective_weights, stock_returns)
pd.testing.assert_series_equal(result.nav, pd.Series([1.1, 1.1, 1.32], index=dates))
pd.testing.assert_frame_equal(result.weights, weights)
pd.testing.assert_frame_equal(result.weights, effective_weights)
@pytest.mark.parametrize("top_k", [0, -1])
+267
View File
@@ -0,0 +1,267 @@
"""No-lookahead factor-score to execution-audit integration contracts."""
from __future__ import annotations
import pandas as pd
import pytest
from quant_engine.execution import ExecutionConfig
from quant_engine.research_pipeline import (
FactorBacktestResult,
FactorExecutionResult,
TargetWeightSchedule,
run_factor_backtest_research,
run_factor_execution_research,
schedule_target_weights,
)
def _calendar() -> pd.DatetimeIndex:
return pd.date_range("2026-01-05", periods=4, freq="B")
def _factor_scores() -> pd.DataFrame:
dates = _calendar()
return pd.DataFrame(
{"A": [2.0, 0.0], "B": [1.0, 3.0]},
index=dates[:2],
)
def _next_session_open_prices() -> pd.DataFrame:
dates = _calendar()
return pd.DataFrame(
{"A": [1.0, 10.0, 10.0, 10.0], "B": [1.0, 10.0, 20.0, 20.0]},
index=dates,
)
def test_schedule_target_weights_maps_signal_to_next_trading_session() -> None:
dates = _calendar()
decision_weights = pd.DataFrame(
{"A": [1.0, 0.0], "B": [0.0, 1.0]},
index=dates[:2],
)
schedule = schedule_target_weights(decision_weights, dates, lag_sessions=1)
assert isinstance(schedule, TargetWeightSchedule)
assert schedule.lag_sessions == 1
pd.testing.assert_series_equal(
schedule.signal_to_execution,
pd.Series(dates[1:3], index=dates[:2], name="execution_date"),
)
expected = decision_weights.copy()
expected.index = dates[1:3]
expected.index.name = "execution_date"
pd.testing.assert_frame_equal(schedule.execution_weights, expected)
assert (schedule.execution_weights.index > schedule.signal_to_execution.index).all()
def test_factor_execution_research_uses_next_session_prices() -> None:
config = ExecutionConfig(
commission_bps=0,
stamp_tax_bps=0,
slippage_bps=0,
min_trade_amount=0,
)
result = run_factor_execution_research(
_factor_scores(),
_next_session_open_prices(),
top_k=1,
execution_price_field="open",
initial_cash=1_000.0,
config=config,
)
assert isinstance(result, FactorExecutionResult)
assert result.execution_price_field == "open"
assert result.execution.daily_executions[0].date == str(_calendar()[1])
assert result.execution.positions[0].holdings == {"A": 100.0}
assert result.execution.positions[1].holdings == {"B": 50.0}
assert result.execution.final_portfolio_value == pytest.approx(1_000.0)
def test_factor_execution_result_snapshots_research_inputs() -> None:
scores = _factor_scores()
prices = _next_session_open_prices()
result = run_factor_execution_research(
scores,
prices,
top_k=1,
execution_price_field="open",
)
scores.iloc[0, 0] = -999.0
prices.iloc[1, 0] = 999.0
assert result.factor_scores.iloc[0, 0] == 2.0
assert result.execution_prices.loc[_calendar()[1], "A"] == 10.0
assert result.execution.positions[0].holdings["A"] < 200_000.0
@pytest.mark.parametrize("lag_sessions", [0, -1, True])
def test_schedule_target_weights_requires_positive_integer_lag(lag_sessions: int) -> None:
with pytest.raises(ValueError, match="lag_sessions"):
schedule_target_weights(
pd.DataFrame({"A": [1.0]}, index=_calendar()[:1]),
_calendar(),
lag_sessions=lag_sessions,
)
def test_schedule_target_weights_rejects_signal_outside_trading_calendar() -> None:
weekend = pd.Timestamp("2026-01-10")
with pytest.raises(ValueError, match="signal dates must be trading sessions"):
schedule_target_weights(
pd.DataFrame({"A": [1.0]}, index=[weekend]),
_calendar(),
)
def test_schedule_target_weights_rejects_missing_future_execution_session() -> None:
dates = _calendar()
with pytest.raises(ValueError, match="future execution session"):
schedule_target_weights(
pd.DataFrame({"A": [1.0]}, index=dates[-1:]),
dates,
)
def test_factor_execution_research_requires_explicit_price_field() -> None:
with pytest.raises(ValueError, match="execution_price_field"):
run_factor_execution_research(
_factor_scores(),
_next_session_open_prices(),
top_k=1,
execution_price_field="",
)
def test_factor_execution_research_accepts_empty_scores() -> None:
scores = pd.DataFrame(columns=["A", "B"], index=pd.DatetimeIndex([]), dtype=float)
result = run_factor_execution_research(
scores,
_next_session_open_prices(),
top_k=1,
execution_price_field="open",
)
assert result.schedule.execution_weights.empty
assert result.execution.positions == ()
def test_factor_backtest_research_runs_signal_to_daily_performance_without_lookahead() -> None:
"""信号日保持现金,下一日开盘成交后才参与当日收盘收益。"""
dates = _calendar()
scores = pd.DataFrame({"A": [2.0], "B": [1.0]}, index=dates[:1])
opens = pd.DataFrame(
{"A": [1.0, 10.0, 10.0, 10.0], "B": [1.0, 20.0, 20.0, 20.0]},
index=dates,
)
closes = pd.DataFrame(
{"A": [500.0, 11.0, 12.0, 12.0], "B": [500.0, 20.0, 20.0, 20.0]},
index=dates,
)
config = ExecutionConfig(
commission_bps=0,
stamp_tax_bps=0,
slippage_bps=0,
min_trade_amount=0,
)
result = run_factor_backtest_research(
scores,
execution_prices=opens,
valuation_prices=closes,
top_k=1,
execution_price_field="open",
valuation_price_field="close",
initial_cash=1_000.0,
config=config,
)
assert isinstance(result, FactorBacktestResult)
assert result.execution_price_field == "open"
assert result.valuation_price_field == "close"
pd.testing.assert_series_equal(
result.nav,
pd.Series([1.0, 1.1, 1.2, 1.2], index=dates, name="nav"),
)
pd.testing.assert_series_equal(
result.returns,
pd.Series([0.0, 0.1, 1.2 / 1.1 - 1.0, 0.0], index=dates, name="returns"),
)
assert result.stats()["n_days"] == 4
assert result.execution.daily_executions[0].executions == ()
assert result.execution.daily_executions[1].executions[0].price == 10.0
def test_factor_backtest_result_snapshots_both_price_semantics() -> None:
scores = pd.DataFrame({"A": [1.0]}, index=_calendar()[:1])
opens = pd.DataFrame({"A": [10.0, 10.0, 10.0, 10.0]}, index=_calendar())
closes = pd.DataFrame({"A": [10.0, 11.0, 12.0, 13.0]}, index=_calendar())
result = run_factor_backtest_research(
scores,
execution_prices=opens,
valuation_prices=closes,
top_k=1,
execution_price_field="open",
valuation_price_field="close",
)
opens.iloc[1, 0] = 999.0
closes.iloc[1, 0] = 999.0
assert result.execution_prices.iloc[1, 0] == 10.0
assert result.valuation_prices.iloc[1, 0] == 11.0
def test_factor_backtest_research_requires_matching_daily_calendars() -> None:
scores = pd.DataFrame({"A": [1.0]}, index=_calendar()[:1])
opens = pd.DataFrame({"A": [10.0, 10.0, 10.0, 10.0]}, index=_calendar())
closes = pd.DataFrame({"A": [10.0, 11.0, 12.0]}, index=_calendar()[:3])
with pytest.raises(ValueError, match="matching trading calendars"):
run_factor_backtest_research(
scores,
execution_prices=opens,
valuation_prices=closes,
top_k=1,
execution_price_field="open",
valuation_price_field="close",
)
def test_factor_backtest_starts_at_first_signal_instead_of_price_warmup() -> None:
"""因子预热行情不能作为空仓日混入研究绩效区间。"""
dates = pd.date_range("2026-01-05", periods=5, freq="B")
scores = pd.DataFrame({"A": [1.0]}, index=dates[2:3])
opens = pd.DataFrame({"A": [1.0, 1.0, 1.0, 10.0, 10.0]}, index=dates)
closes = pd.DataFrame({"A": [100.0, 200.0, 300.0, 11.0, 12.0]}, index=dates)
config = ExecutionConfig(
commission_bps=0,
stamp_tax_bps=0,
slippage_bps=0,
min_trade_amount=0,
)
result = run_factor_backtest_research(
scores,
execution_prices=opens,
valuation_prices=closes,
top_k=1,
execution_price_field="open",
valuation_price_field="close",
initial_cash=1_000.0,
config=config,
)
assert result.nav.index.equals(dates[2:])
pd.testing.assert_series_equal(
result.nav,
pd.Series([1.0, 1.1, 1.2], index=dates[2:], name="nav"),
)
assert result.stats()["n_days"] == 3