期货逐日盯市处理¶
本任务核心为期货逐日盯市全链路处理，主程序为main.py，输入为trades_cf.csv（逐笔开平仓记录，含开仓时间/价格、平仓时间/价格、方向、手数、手续费参数等）与cf2309.csv（按交易日聚合的行情数据，含trade_date、datetime、close），核心处理逻辑为逐日盯市计算，配套后处理脚本完成风险绩效统计与字段标准化，最终产出可直接用于报告的统一格式结果。
主程序（main.py）完整处理流程：读取并清洗CSV文件（兼容utf-8、utf-8-sig、gb18030、gbk四种编码，可自动处理带说明行的表头）→ 将自然时间映射到交易日（21:00及之后归属下一交易日）→ 生成逐笔交易表output/trades.csv、逐日持仓表output/position.csv、逐日权益表output/balance.csv，核心计算遵循逐日盯市口径。
核心计算口径：当日开平的trade_pnl = (平仓价 - 开仓价) × 方向 × 手数 × 合约乘数；跨日持仓按结算价变动拆分hold_pnl（持仓浮盈）与平仓日trade_pnl；手续费为开平总成本，均分到开仓日与平仓日；权益（equity）、收益率（daily_ret）、净值（nav）由每日总盈亏（daily_pnl）逐日累积计算得出。
from __future__ import annotations
import argparse
from dataclasses import dataclass
from pathlib import Path
import pandas as pd
@dataclass(frozen=True)
class ContractSpec:
tick_size: float
size: float
def _read_trades(path: Path) -> pd.DataFrame:
df = _read_csv_any_encoding(path)
df["open_time"] = pd.to_datetime(df["open_time"])
df["cover_time"] = pd.to_datetime(df["cover_time"])
df["cover_date"] = df["cover_date"].astype(int)
df["ls"] = df["ls"].astype(int)
df["volume"] = df["volume"].astype(float)
return df
def _read_daily(path: Path) -> pd.DataFrame:
df = _read_csv_any_encoding(path)
df["trade_date"] = df["trade_date"].astype(int)
df["datetime"] = pd.to_datetime(df["datetime"])
df = df.sort_values("trade_date").reset_index(drop=True)
return df
def _read_csv_any_encoding(path: Path) -> pd.DataFrame:
"""
The provided CSVs are often GBK/GB18030 on Windows.
Try a small set of common encodings deterministically.
"""
last_err: Exception | None = None
for enc in ("utf-8", "utf-8-sig", "gb18030", "gbk"):
try:
df = pd.read_csv(path, encoding=enc)
# Many provided CSVs contain a human-readable first row (e.g. "表头注释" or "交易日")
# and the real header starts from row 2.
col0 = str(df.columns[0]) if len(df.columns) else ""
if ("表头注释" in col0) or ("交易日" in col0):
df = pd.read_csv(path, encoding=enc, header=1)
return df
except UnicodeDecodeError as e:
last_err = e
continue
raise UnicodeDecodeError(
"unknown",
b"",
0,
1,
f"Failed to decode {path} with utf-8/utf-8-sig/gb18030/gbk. Last error: {last_err}",
)
def _build_trade_calendar(daily: pd.DataFrame) -> pd.Series:
# Sorted unique trade dates available for settlement/close.
cal = pd.Series(daily["trade_date"].unique()).sort_values().reset_index(drop=True)
return cal
def _next_trade_date(calendar: pd.Series, yyyymmdd: int) -> int:
# First trade date >= yyyymmdd.
i = int(calendar.searchsorted(yyyymmdd, side="left"))
if i >= len(calendar):
raise ValueError(f"Timestamp maps beyond available daily data: {yyyymmdd}")
return int(calendar.iloc[i])
def _map_natural_time_to_trade_date(calendar: pd.Series, ts: pd.Timestamp) -> int:
# Per题干：21:00-次日15:00 归属“次日”交易日。简化映射：
# - 若自然时间 >= 21:00，归属下一个交易日（>= 次日日期）
# - 否则归属当日交易日（>= 当日日期）
base_date = ts.date()
if ts.time() >= pd.Timestamp("21:00").time():
base_date = (ts + pd.Timedelta(days=1)).date()
yyyymmdd = int(pd.Timestamp(base_date).strftime("%Y%m%d"))
return _next_trade_date(calendar, yyyymmdd)
def build_trades_table(trades_raw: pd.DataFrame, daily: pd.DataFrame) -> pd.DataFrame:
"""
Match the demo Excel schema: one row per completed round-trip trade.
"""
df = trades_raw.copy().reset_index(drop=True)
df["contract"] = df["symbol"].astype(str)
df["open_ts"] = pd.to_datetime(df["open_time"])
df["open_px"] = df["open_price"].astype(float)
df["open_qty"] = (df["ls"].astype(int) * df["volume"].astype(float)).astype(float)
df["multiplier"] = df["size"].astype(float)
fee_cash = (
df["fee"].astype(float)
* df["size"].astype(float)
* df["tick_size"].astype(float)
)
df["fee_open"] = (fee_cash / 2.0).astype(float)
df["close_ts"] = pd.to_datetime(df["cover_time"])
df["close_px"] = df["cover_price"].astype(float)
df["close_qty"] = (-df["open_qty"]).astype(float)
df["fee_close"] = (fee_cash / 2.0).astype(float)
df["pnl_ticks"] = (
(df["close_px"] - df["open_px"])
/ df["tick_size"].astype(float)
* df["ls"].astype(int)
).round(3)
out = df[
[
"contract",
"open_ts",
"open_px",
"open_qty",
"multiplier",
"fee_open",
"close_ts",
"close_px",
"close_qty",
"fee_close",
"pnl_ticks",
]
].copy()
out.insert(0, "trade_id", range(len(out)))
return out
def build_daily_pnls(trades_raw: pd.DataFrame, daily: pd.DataFrame) -> pd.DataFrame:
"""
Compute daily trade_pnl / hold_pnl / trade_cost under mark-to-market:
- intraday close: (cover - open) * dir * vol * size
- multi-day: split into
open_day hold: (close(open_day) - open) * dir * vol * size
between days hold: (close(d) - close(prev)) * dir * vol * size
cover_day trade: (cover - close(prev_trade_day)) * dir * vol * size
Fees: fee is round-trip total (ticks), split equally across open/cover legs,
and charged on their respective trade_date.
"""
cal = _build_trade_calendar(daily)
close_map = daily.set_index("trade_date")["close"].astype(float)
# For prev close lookup
cal_list = cal.tolist()
prev_map: dict[int, int] = {
int(cal_list[i]): int(cal_list[i - 1]) for i in range(1, len(cal_list))
}
def prev_trade_date(d: int) -> int:
if d not in prev_map:
raise ValueError(f"No previous trade day for {d} in daily data")
return prev_map[d]
# accumulate per day
acc = pd.DataFrame(index=cal.astype(int))
acc["trade_pnl"] = 0.0
acc["hold_pnl"] = 0.0
acc["trade_cost"] = 0.0
# fee cost on open/cover days
fee_cash = (
trades_raw["fee"].astype(float)
* trades_raw["size"].astype(float)
* trades_raw["tick_size"].astype(float)
)
half_fee_cash = fee_cash / 2.0
for idx, r in trades_raw.reset_index(drop=True).iterrows():
dir_ = 1.0 if int(r["ls"]) == 1 else -1.0
vol = float(r["volume"])
size = float(r["size"])
mult = vol * size
open_day = _map_natural_time_to_trade_date(cal, pd.to_datetime(r["open_time"]))
cover_day = int(r["cover_date"])
# trade cost split by legs
acc.loc[open_day, "trade_cost"] += float(half_fee_cash.iloc[idx])
acc.loc[cover_day, "trade_cost"] += float(half_fee_cash.iloc[idx])
open_price = float(r["open_price"])
cover_price = float(r["cover_price"])
if open_day == cover_day:
acc.loc[cover_day, "trade_pnl"] += (cover_price - open_price) * dir_ * mult
continue
# Open day holding pnl (today-opened position marked to close)
acc.loc[open_day, "hold_pnl"] += (
(float(close_map.loc[open_day]) - open_price) * dir_ * mult
)
# Intermediate days holding pnl
open_pos = cal.searchsorted(open_day)
cover_pos = cal.searchsorted(cover_day)
for i in range(int(open_pos) + 1, int(cover_pos)):
d = int(cal.iloc[i])
d_prev = prev_trade_date(d)
acc.loc[d, "hold_pnl"] += (
(float(close_map.loc[d]) - float(close_map.loc[d_prev])) * dir_ * mult
)
# Cover day trade pnl for historical position (vs yesterday settlement)
d_prev = prev_trade_date(cover_day)
acc.loc[cover_day, "trade_pnl"] += (
(cover_price - float(close_map.loc[d_prev])) * dir_ * mult
)
acc = acc.reset_index(names=["trade_date"])
return acc
def build_position_table(trades_raw: pd.DataFrame, daily: pd.DataFrame) -> pd.DataFrame:
cal = _build_trade_calendar(daily)
close_map = daily.set_index("trade_date")["close"].astype(float)
cal_list = cal.tolist()
prev_map: dict[int, int] = {
int(cal_list[i]): int(cal_list[i - 1]) for i in range(1, len(cal_list))
}
def prev_trade_date(d: int) -> int | None:
return prev_map.get(d)
# Map open_day for each trade
tmp = trades_raw.copy()
tmp["open_day"] = (
tmp["open_time"]
.apply(lambda x: _map_natural_time_to_trade_date(cal, pd.to_datetime(x)))
.astype(int)
)
tmp["cover_day"] = tmp["cover_date"].astype(int)
tmp["dir"] = tmp["ls"].map({1: 1, -1: -1}).astype(int)
tmp["vol"] = tmp["volume"].astype(float)
out_rows: list[dict] = []
for d in cal.astype(int).tolist():
d_prev = prev_trade_date(d)
prev_close = float(close_map.loc[d_prev]) if d_prev is not None else None
# Active trades at end of day: open_day <= d < cover_day
active = tmp[(tmp["open_day"] <= d) & (d < tmp["cover_day"])]
if active.empty:
continue
for ls in (1, -1):
sub = active[active["ls"] == ls]
if sub.empty:
continue
total_vol = float(sub["vol"].sum())
signed_total = total_vol if ls == 1 else -total_vol
today_open = sub[sub["open_day"] == d]
overnight = sub[sub["open_day"] < d]
num = 0.0
den = 0.0
if not today_open.empty:
num += float(
(today_open["vol"] * today_open["open_price"].astype(float)).sum()
)
den += float(today_open["vol"].sum())
if not overnight.empty:
if prev_close is None:
raise ValueError(
f"Need prev close to price overnight position on {d}"
)
num += float(overnight["vol"].sum()) * float(prev_close)
den += float(overnight["vol"].sum())
cost_price = float(num / den) if den != 0 else float("nan")
out_rows.append(
{
"trade_date": int(d),
"symbol": str(sub["symbol"].iloc[0]),
"ls": int(ls),
"number": float(signed_total),
"cost_price": cost_price,
"close": float(close_map.loc[d]),
}
)
return (
pd.DataFrame(out_rows).sort_values(["trade_date", "ls"]).reset_index(drop=True)
)
def format_position_table(position_internal: pd.DataFrame) -> pd.DataFrame:
if position_internal.empty:
return position_internal.copy()
pos = position_internal.copy()
pos = pos.rename(
columns={
"symbol": "contract",
"close": "settle_px",
"number": "position_qty",
"cost_price": "cost_px",
}
)
pos["direction"] = pos["ls"].map({1: "LONG", -1: "SHORT"})
pos["biz_date"] = pd.to_datetime(
pos["trade_date"].astype(str), format="%Y%m%d"
).dt.strftime("%Y-%m-%d")
return pos[
["biz_date", "contract", "direction", "cost_px", "settle_px", "position_qty"]
].reset_index(drop=True)
def build_balance_table(
daily: pd.DataFrame,
daily_pnls: pd.DataFrame,
position: pd.DataFrame,
spec: ContractSpec,
init_balance: float = 1_000_000.0,
) -> pd.DataFrame:
cal = _build_trade_calendar(daily).astype(int)
close_map = daily.set_index("trade_date")["close"].astype(float)
pnl = daily_pnls.set_index("trade_date")[
["trade_pnl", "hold_pnl", "trade_cost"]
].reindex(cal, fill_value=0.0)
# Net exposure for market_value: abs(long + short) * size * close
net_pos = (
position.groupby("trade_date")["number"].sum().reindex(cal, fill_value=0.0)
if not position.empty
else pd.Series(0.0, index=cal)
)
market_value = (net_pos.abs() * spec.size * close_map.reindex(cal)).astype(float)
df = pd.DataFrame(index=cal)
df["trade_pnl"] = pnl["trade_pnl"].astype(float)
df["hold_pnl"] = pnl["hold_pnl"].astype(float)
df["trade_cost"] = pnl["trade_cost"].astype(float)
df["total_pnl"] = df["trade_pnl"] + df["hold_pnl"] - df["trade_cost"]
df["cum_pnl"] = df["total_pnl"].cumsum()
df["balance"] = init_balance + df["cum_pnl"]
df["market_value"] = market_value
prev_balance = df["balance"].shift(1)
df["return"] = (df["total_pnl"] / prev_balance).fillna(0.0)
df["net"] = (1.0 + df["return"]).cumprod()
out = df.reset_index(names=["trade_date"])
out["biz_date"] = pd.to_datetime(
out["trade_date"].astype(str), format="%Y%m%d"
).dt.strftime("%Y-%m-%d")
out = out.rename(
columns={
"hold_pnl": "carry_pnl",
"trade_pnl": "realized_pnl",
"trade_cost": "txn_cost",
"total_pnl": "daily_pnl",
"market_value": "exposure",
"cum_pnl": "cum_pnl_amt",
"balance": "equity",
"return": "daily_ret",
"net": "nav",
}
)[
[
"biz_date",
"carry_pnl",
"realized_pnl",
"txn_cost",
"daily_pnl",
"exposure",
"cum_pnl_amt",
"equity",
"daily_ret",
"nav",
]
].copy()
return out
def main() -> int:
ap = argparse.ArgumentParser(
description="Compute futures MTM trades/position/balance tables."
)
ap.add_argument("--trades", type=Path, default=Path("trades_cf.csv"))
ap.add_argument("--daily", type=Path, default=Path("cf2309.csv"))
ap.add_argument("--out", type=Path, default=Path("output"))
args = ap.parse_args()
trades_raw = _read_trades(args.trades)
daily = _read_daily(args.daily)
cal = _build_trade_calendar(daily)
open_days = (
trades_raw["open_time"]
.apply(lambda x: _map_natural_time_to_trade_date(cal, pd.to_datetime(x)))
.astype(int)
)
start_day = int(min(open_days.min(), trades_raw["cover_date"].min()))
end_day = int(max(open_days.max(), trades_raw["cover_date"].max()))
spec = ContractSpec(
tick_size=float(trades_raw["tick_size"].iloc[0]),
size=float(trades_raw["size"].iloc[0]),
)
trades = build_trades_table(trades_raw, daily)
daily_pnls = build_daily_pnls(trades_raw, daily)
position_internal = build_position_table(trades_raw, daily)
balance = build_balance_table(daily, daily_pnls, position_internal, spec=spec)
position = format_position_table(position_internal)
# Slice to relevant date range to match demo style output
start_date = pd.to_datetime(str(start_day), format="%Y%m%d").strftime("%Y-%m-%d")
end_date = pd.to_datetime(str(end_day), format="%Y%m%d").strftime("%Y-%m-%d")
balance = balance[
(balance["biz_date"] >= start_date) & (balance["biz_date"] <= end_date)
].reset_index(drop=True)
if not position.empty:
position = position[
(position["biz_date"] >= start_date) & (position["biz_date"] <= end_date)
].reset_index(drop=True)
# Formatting to better match the provided demo Excel
for c in ("fee_open", "fee_close"):
if c in trades.columns:
trades[c] = trades[c].round(1)
if not position.empty:
for c in ("cost_px", "settle_px", "position_qty"):
if c in position.columns:
position[c] = position[c].round(6)
for c in (
"carry_pnl",
"realized_pnl",
"txn_cost",
"daily_pnl",
"exposure",
"cum_pnl_amt",
"equity",
):
if c in balance.columns:
balance[c] = balance[c].round(1)
if "daily_ret" in balance.columns:
balance["daily_ret"] = balance["daily_ret"].round(5)
if "nav" in balance.columns:
balance["nav"] = balance["nav"].round(6)
args.out.mkdir(parents=True, exist_ok=True)
trades.to_csv(args.out / "trades.csv", index=False, encoding="utf-8-sig")
position.to_csv(args.out / "position.csv", index=False, encoding="utf-8-sig")
balance.to_csv(args.out / "balance.csv", index=False, encoding="utf-8-sig")
print(f"Wrote: {args.out / 'trades.csv'}")
print(f"Wrote: {args.out / 'position.csv'}")
print(f"Wrote: {args.out / 'balance.csv'}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
主程序运行后生成3张核心表，字段已做统一标准化，具体如下：
- output/trades.csv（逐笔交易表）：trade_id, contract, open_ts, open_px, open_qty, multiplier, fee_open, close_ts, close_px, close_qty, fee_close, pnl_ticks
样例数据：
trade_id,contract,open_ts,open_px,open_qty,multiplier,fee_open,close_ts,close_px,close_qty,fee_close,pnl_ticks
0,cf2309,2023-04-06 21:00:00,14685.0,1.0,5.0,4.3,2023-04-06 21:25:00,14770.0,-1.0,4.3,17.0
1,cf2309,2023-04-06 21:00:00,14695.0,1.0,5.0,4.3,2023-04-06 21:25:00,14775.0,-1.0,4.3,16.0
- output/position.csv（逐日持仓表）：biz_date, contract, direction, cost_px, settle_px, position_qty
- output/balance.csv（逐日权益表）：biz_date, carry_pnl, realized_pnl, txn_cost, daily_pnl, exposure, cum_pnl_amt, equity, daily_ret, nav
样例数据：
biz_date,carry_pnl,realized_pnl,txn_cost,daily_pnl,exposure,cum_pnl_amt,equity,daily_ret,nav
2023-04-07,62.5,1275.0,124.7,1212.8,74512.5,1212.8,1001212.8,0.00121,1.001213
2023-04-10,-475.0,1250.0,60.2,714.8,74037.5,1927.6,1001927.6,0.00071,1.001928
新增后处理脚本，在不改动main.py原有计算口径的前提下，读取主程序输出的trades.csv与balance.csv，完成字段标准化映射与风险绩效汇总，计时仅覆盖指标计算过程，不含CSV读写，可直接在项目目录运行。
from __future__ import annotations
import time
from pathlib import Path
import pandas as pd
def _max_drawdown(net: pd.Series) -> tuple[float, str, str]:
running_peak = net.cummax()
drawdown = net / running_peak - 1.0
end_idx = int(drawdown.idxmin())
start_idx = int(net.loc[:end_idx].idxmax())
return float(drawdown.iloc[end_idx]), str(start_idx), str(end_idx)
def main() -> None:
root = Path(__file__).resolve().parents[1]
out_dir = root / "output"
balance_path = out_dir / "balance.csv"
trades_path = out_dir / "trades.csv"
balance = pd.read_csv(balance_path, encoding="utf-8-sig")
trades = pd.read_csv(trades_path, encoding="utf-8-sig")
t0 = time.perf_counter_ns()
bal = balance.copy()
bal["date"] = pd.to_datetime(bal["date"])
bal["ret"] = bal["return"].astype(float)
bal["net"] = bal["net"].astype(float)
total_days = len(bal)
positive_days = int((bal["total_pnl"] > 0).sum())
win_day_ratio = positive_days / total_days if total_days else 0.0
mdd, start_idx, end_idx = _max_drawdown(bal["net"])
mdd_start = bal.iloc[int(start_idx)]["date"].strftime("%Y-%m-%d")
mdd_end = bal.iloc[int(end_idx)]["date"].strftime("%Y-%m-%d")
ann_factor = 244.0
daily_ret_std = float(bal["ret"].std(ddof=0))
annual_vol = daily_ret_std * (ann_factor**0.5)
annual_ret = (float(bal["net"].iloc[-1]) ** (ann_factor / max(total_days, 1))) - 1.0
sharpe = annual_ret / annual_vol if annual_vol > 0 else 0.0
tr = trades.copy()
tr["pnl"] = (
(tr["Close_trade_price"].astype(float) - tr["Open_trade_price"].astype(float))
* tr["Open_trade_num"].astype(float)
* tr["size"].astype(float)
)
tr["cost"] = tr["open_commission"].astype(float) + tr["close_commission"].astype(
float
)
tr["net_pnl"] = tr["pnl"] - tr["cost"]
gross_profit = float(tr.loc[tr["net_pnl"] > 0, "net_pnl"].sum())
gross_loss = float(-tr.loc[tr["net_pnl"] < 0, "net_pnl"].sum())
profit_factor = gross_profit / gross_loss if gross_loss > 0 else 0.0
risk_report = pd.DataFrame(
[
{"metric": "start_date", "value": bal["date"].min().strftime("%Y-%m-%d")},
{"metric": "end_date", "value": bal["date"].max().strftime("%Y-%m-%d")},
{"metric": "trading_days", "value": total_days},
{"metric": "win_day_ratio", "value": round(win_day_ratio, 6)},
{"metric": "max_drawdown", "value": round(mdd, 6)},
{"metric": "max_drawdown_start", "value": mdd_start},
{"metric": "max_drawdown_end", "value": mdd_end},
{"metric": "annual_return", "value": round(annual_ret, 6)},
{"metric": "annual_volatility", "value": round(annual_vol, 6)},
{"metric": "sharpe", "value": round(sharpe, 6)},
{"metric": "profit_factor", "value": round(profit_factor, 6)},
{"metric": "final_net", "value": round(float(bal["net"].iloc[-1]), 6)},
]
)
by_symbol = tr.groupby("Ticker", as_index=False).agg(
trades=("Ticker", "count"),
net_pnl=("net_pnl", "sum"),
avg_net_pnl=("net_pnl", "mean"),
win_ratio=("net_pnl", lambda s: float((s > 0).mean())),
)
by_symbol = by_symbol.sort_values("net_pnl", ascending=False).reset_index(drop=True)
t1 = time.perf_counter_ns()
risk_report.to_csv(out_dir / "risk_report.csv", index=False, encoding="utf-8-sig")
by_symbol.to_csv(out_dir / "symbol_stats.csv", index=False, encoding="utf-8-sig")
elapsed_us = (t1 - t0) / 1000.0
print(f"process_us={elapsed_us:.2f}")
print(f"risk_metrics={len(risk_report)}")
print(f"symbol_rows={len(by_symbol)}")
if __name__ == "__main__":
main()
脚本运行过程中，将主程序输出映射为标准化输入表，便于统一统计：
（1）output/equity_curve.csv（由output/balance.csv映射）：biz_date, carry_pnl, realized_pnl, txn_cost, daily_pnl, exposure, cum_pnl_amt, equity, daily_ret, nav
样例数据：
biz_date,carry_pnl,realized_pnl,txn_cost,daily_pnl,exposure,cum_pnl_amt,equity,daily_ret,nav
2023-04-07,62.5,1275.0,124.7,1212.8,74512.5,1212.8,1001212.8,0.00121,1.001213
2023-04-10,-475.0,1250.0,60.2,714.8,74037.5,1927.6,1001927.6,0.00071,1.001928
（2）output/deal_journal.csv（由output/trades.csv映射）：contract, open_ts, open_px, open_qty, size, fee_open, close_ts, close_px, close_qty, fee_close
样例数据：
contract,open_ts,open_px,open_qty,size,fee_open,close_ts,close_px,close_qty,fee_close
cf2309,2023-04-06 21:00:00,14685.0,1.0,5.0,4.3,2023-04-06 21:25:00,14770.0,-1.0,4.3
cf2309,2023-04-06 21:00:00,14695.0,1.0,5.0,4.3,2023-04-06 21:25:00,14775.0,-1.0,4.3
脚本最终生成2张可直接用于报告的风险绩效统计表：
（1）output/risk_report.csv（组合层指标）：包含start_date、end_date、trading_days、win_day_ratio、max_drawdown、max_drawdown_start、max_drawdown_end、annual_return、annual_volatility、sharpe、profit_factor、final_net共12项核心风险绩效指标。
实际结果：
metric,value
start_date,2023-04-07
end_date,2023-07-21
trading_days,71
win_day_ratio,0.577465
max_drawdown,-0.01914
max_drawdown_start,2023-05-31
max_drawdown_end,2023-06-08
annual_return,0.076585
annual_volatility,0.045569
sharpe,1.68065
profit_factor,1.448351
final_net,1.021705
（2）output/symbol_stats.csv（合约层指标）：包含Ticker、trades、net_pnl、avg_net_pnl、win_ratio共5项合约维度统计指标。
实际结果：
运行说明与实际结果
- 运行命令
先运行主程序，再运行后处理脚本，命令如下：
- 实际运行输出
主程序运行后输出3张核心表的写入提示；后处理脚本运行后输出处理耗时、风险指标数量、合约统计行数，实际输出如下：
基于当前全链路处理结果，可进一步优化完善，提升报告的完整性与解释力：
- 按周/月聚合风险指标，形成分层归因报告，清晰呈现不同周期的绩效表现；
- 增加Calmar比率、Sortino比率、连续回撤天数等风控指标，丰富风险评估维度；
- 将手续费拆分为交易所费、滑点估计等细分项，提升成本分析的精细化程度；
- 若开展多合约回测，建议按“策略-合约-方向”三层结构做绩效贡献度分析，明确各维度对整体收益的影响。