第 46 篇：从模拟盘到实盘纸面交易与切换清单

# ---------- 1. 纸面交易引擎（承接第45篇 PaperBroker，精简内联） ----------
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path

@dataclass
class Order:
    symbol: str; side: str; price: float; volume: int
    status: str = "pending"; filled_volume: int = 0
    msgs: list = field(default_factory=list)

class PaperBroker:
    def __init__(self, init_cash=100000.0, fee=0.001, slip=0.001):
        self.cash = init_cash
        self.pos = {}
        self.fee, self.slip = fee, slip
        self.orders = []                      # 全部委托留痕

    def _fill(self, symbol, side, price, volume):
        fill_px = price * (1 + self.slip if side == "buy" else -self.slip)
        cost = volume * fill_px * (1 + self.fee if side == "buy" else -self.fee)
        self.cash -= cost
        self.pos[symbol] = self.pos.get(symbol, 0) + \
            (volume if side == "buy" else -volume)
        o = Order(symbol, side, price, volume, "filled", volume,
                  [f"{datetime.now():%H:%M:%S} 成交价{fill_px:.3f}"])
        self.orders.append(o)
        return o

    def buy(self, symbol, price, volume):
        return self._fill(symbol, "buy", price, volume)

    def sell(self, symbol, price, volume):
        return self._fill(symbol, "sell", price, volume)

    def position(self, symbol):
        return self.pos.get(symbol, 0)

# ---------- 2. 每日信号 → 委托 → 记录（纸面主循环的单日切片） ----------
def get_market_data(symbol):
    """获取当日行情。akshare 优先，失败用模拟价（教学）"""
    try:
        import akshare as ak
        df = ak.fund_etf_hist_em(symbol=symbol, period="daily")
        row = df.iloc[-1]
        return float(row["收盘"])
    except Exception:
        rng = np.random.default_rng(int(datetime.now().strftime("%H%M")))
        return 4.0 * (1 + rng.normal(0, 0.01))     # 模拟当日收盘价

def daily_signal(px_history):
    """双均线信号：返回 1(持有多头) / 0(空仓)。只用历史数据"""
    s = pd.Series(px_history)
    return int(s.rolling(5).mean().iloc[-1] > s.rolling(20).mean().iloc[-1])

def run_paper_day(broker, symbol, px_history, trade_log):
    """纸面交易的完整单日流程"""
    price = get_market_data(symbol)
    sig = daily_signal(px_history)           # ① 收盘后算信号
    cur = broker.position(symbol)
    want = sig * 10000                        # 目标持仓股数
    if want > cur:                            # ② 生成委托（次日口径简化为当日）
        o = broker.buy(symbol, price, want - cur)
    elif want < cur:
        o = broker.sell(symbol, price, cur - want)
    else:
        o = None
    # ③ 记录：日期/信号/价格/动作/现金/持仓
    trade_log.append({
        "date": datetime.now().strftime("%Y-%m-%d"),
        "signal": sig, "price": round(price, 3),
        "action": (o.side if o else "hold"),
        "volume": (o.volume if o else 0),
        "cash": round(broker.cash, 2),
        "position": broker.position(symbol),
    })
    return trade_log

# ---------- 3. 每日复盘：净值跟踪与信号核对 ----------
def paper_review(broker, symbol, trade_log):
    """收盘后复盘：当前净值、当日动作、累计交易统计"""
    price = get_market_data(symbol)
    nav = broker.cash + broker.position(symbol) * price
    df = pd.DataFrame(trade_log)
    n_switch = (df["action"] != "hold").sum() if len(df) else 0
    print(f"--- 纸面复盘 {datetime.now():%Y-%m-%d} ---")
    print(f"当前净值: {nav:,.2f} (初始 100,000)")
    print(f"累计信号切换: {n_switch} 次, 委托总数: {len(broker.orders)}")
    if len(df):
        print("最近3天动作:")
        print(df.tail(3).to_string(index=False))
    Path("logs").mkdir(exist_ok=True)
    df.to_csv("logs/paper_trades.csv", index=False, encoding="utf-8-sig")
    return nav

# ---------- 演示运行 ----------
broker = PaperBroker()
rng = np.random.default_rng(46)
px_history = list(4.0 * np.exp(np.cumsum(rng.normal(0.0002, 0.01, 60))))
trade_log = []
for day in range(3):                          # 模拟连续3个交易日
    px_history.append(get_market_data("510300"))
    trade_log = run_paper_day(broker, "510300", px_history, trade_log)
paper_review(broker, "510300", trade_log)

# ---------- 4. 切换实盘前的检查清单评估器 ----------
def readiness_check():
    """逐项评估。任何 NO → 不具备切换条件"""
    checks = [
        ("纸面交易连续运行满 3 个月，无中断", None),
        ("纸面期间信号生成 0 次迟到/缺失", None),
        ("策略参数已冻结（写入 git commit，见第44篇）", None),
        ("实盘通道已开通并完成 1 笔最小单位真实委托", None),
        ("断线重连演练：手动断网后系统能恢复持仓状态", None),
        ("异常预案成文：接口失效/数据缺失/涨跌停的处理动作", None),
        ("实盘资金计划：金额 ≤ 总资产可亏损部分，且 3 个月不动用", None),
        ("风控在线：止损规则(第35篇)在纸面期间至少触发过 1 次", None),
        ("监控告警实测：人为制造一次异常，告警能送达", None),
        ("对账机制：每日策略持仓 vs 通道持仓比对", None),
    ]
    print("=== 实盘切换检查清单 ===")
    for i, (item, _) in enumerate(checks, 1):
        ans = input(f"{i:2d}. {item} [y/n]: ").strip().lower()
        if ans != "y":
            print(f">>> 第 {i} 项未通过 —— 请补齐后再切换实盘")
            return False
    print(">>> 全部通过，可以进入小资金实盘阶段")
    return True

# readiness_check()   # 取消注释交互式运行

示例输出（示意，模拟行情下）:
--- 纸面复盘 2026-09-28 ---
当前净值: 99,842.15 (初始 100,000)
累计信号切换: 2 次, 委托总数: 2
最近3天动作:
        date  signal  price action  volume      cash  position
0  2026-09-24       0  4.052   hold       0  100000.0         0
1  2026-09-25       1  4.061    buy   10000   58627.4     10000
2  2026-09-28       1  4.048   hold       0   58627.4     10000