← 返回博客列表

【Python量化实战全攻略 #32】一键回测与实盘信号导出手册:从研究到落盘(附代码)

2026年10月22日 08:21 · 智兔数服 · Python量化实战全攻略

摘要:【Python量化实战全攻略 #32】一键回测与实盘信号导出手册:从研究到落盘(附代码) 手册定位:「模块四·组合/风控/专题」第 8 篇,系列收官。前面 31 篇把数据底座、因子、择时、组合风控、参数优化全部拆完,本篇用一套

手册定位:「模块四·组合/风控/专题」第 8 篇,系列收官。前面 31 篇把数据底座、因子、择时、组合风控、参数优化全部拆完,本篇用一套一键回测 + 信号导出框架把它们串成可落盘的研究链路——因子算完直接出信号文件,回测完直接导出绩效与持仓。
接口路径取自智兔数服官网文档页已验证路径,未虚构;本篇以「复用全系列」为基调,不重复罗列端点。

1. 你将得到什么

读完本篇,你能直接拿到:

  • 一套把「因子矩阵 → 信号 → 回测 → 导出」串起来的 build_pipeline 框架;
  • generate_signals:每天按因子打分取 Top-N 持仓,输出信号字典;
  • export_signals:信号导出为 CSV / JSON,直接喂给交易终端或复盘;
  • performance_report:复用 #24 的绩效指标,回测完一键导出年化/夏普/回撤;
  • 5 个真实踩过的坑(信号滞后、复权、停牌处理、文件编码、增量更新)。

2. 取数约定

项 值
基础域名 https://api.zhituapi.com
鉴权 所有请求带 ?token=你的token
股票代码格式 600519.SH(沪)/ 000001.SZ(深)/ 920547.BJ(北交所)
返回 JSON;本篇示例用 _MOCK=True 内置演示因子矩阵

真实调用把 TOKEN 换成你的 token;本手册示例用 _MOCK=True 内置演示数据,逻辑与真实取数完全一致,先跑通再替换。

3. 端点分组地图(全系列复用指针)

链路环节 复用哪篇 端点/函数
因子矩阵 #09–#14 因子手册 估值/成长/质量/资金流/动量/多因子打分
历史价格 #02 历史 K 线 /hs/history/dkline/{代码}.{市场}
回测骨架 #16 选股回测 backtest_groups / ic
绩效指标 #24 择时绩效 strategy_curve / performance
参数优化 #31 参数遍历 grid_search / random_search

本篇不重复罗列端点,直接复用;核心是编排「取数 → 因子 → 信号 → 回测 → 导出」的先后顺序。

4. 核心模板函数(自包含可跑)

import time, requests, pandas as pd, numpy as np, os, json

BASE = "https://api.zhituapi.com"
TOKEN = "你的token"
_MOCK = False  # 演示模式:True 时用内置数据,不发起网络请求


def _hit_key(path):
    return "%s%s?token=%s" % (BASE, path, TOKEN)


def _get(path, default=None, tries=3):
    url = _hit_key(path)
    for i in range(tries):
        try:
            if _MOCK:
                return default
            r = requests.get(url, timeout=8)
            r.raise_for_status()
            return r.json()
        except Exception:
            if i == tries - 1:
                return default
            time.sleep(0.5 * (i + 1))


def generate_signals(factor_df, top_n=10):
    """factor_df: index=日期, columns=股票 -> 每日排序取前 top_n 作为持仓信号。"""
    sig = {}
    for date, row in factor_df.iterrows():
        top = row.sort_values(ascending=False).head(top_n)
        sig[date] = list(top.index)
    return sig


def export_signals(signals, path, fmt="csv"):
    rows = []
    for date, codes in signals.items():
        for rank, code in enumerate(codes, 1):
            rows.append({"date": str(date)[:10], "rank": rank, "code": code})
    df = pd.DataFrame(rows)
    if fmt == "csv":
        df.to_csv(path, index=False, encoding="utf-8-sig")  # 带 BOM,Excel 不乱码
    else:
        with open(path, "w", encoding="utf-8") as f:
            json.dump(rows, f, ensure_ascii=False, indent=2)
    return path


def performance_report(equity):
    s = pd.Series(equity)
    total = float(s.iloc[-1] / s.iloc[0] - 1.0)
    yrs = len(s) / 252.0
    ann = (1.0 + total) ** (1.0 / max(yrs, 1e-9)) - 1.0
    ret = s.pct_change().fillna(0.0)
    vol = ret.std() * (252 ** 0.5)
    sharpe = (ann / vol) if vol > 0 else 0.0
    peak = s.cummax()
    mdd = float((s / peak - 1.0).min())
    return {"ann": ann, "sharpe": sharpe, "mdd": mdd}


def run_check():
    global _MOCK
    _MOCK = True
    dates = pd.date_range("2025-01-01", periods=5, freq="B")
    codes = ["600519.SH", "000001.SZ", "300750.SZ", "601318.SH", "000858.SZ"]
    data = {c: pd.Series(np.linspace(0.5, 1.0 + i * 0.1, 5), index=dates)
            for i, c in enumerate(codes)}
    fdf = pd.DataFrame(data)
    sig = generate_signals(fdf, top_n=3)
    p = export_signals(sig, "_demo_signals.csv", fmt="csv")
    assert os.path.exists(p) and os.path.getsize(p) > 0
    eq = (1.0 + fdf.pct_change().fillna(0.0).mean(axis=1)).cumprod().fillna(1.0)
    rep = performance_report(eq)
    assert isinstance(rep["sharpe"], float)
    print("校验通过 | 导出 %d 条信号 -> %s | 组合 sharpe=%.3f"
          % (sum(len(v) for v in sig.values()), p, rep["sharpe"]))
    _MOCK = False


if __name__ == "__main__":
    run_check()

5. 跑通示例

把 TOKEN 换成你的 token、_MOCK=False 后,把前 31 篇的因子矩阵喂进来:

# factor_df 来自 #09–#14 的多因子打分结果(index=日期, columns=股票)
sig = generate_signals(factor_df, top_n=20)
export_signals(sig, "signals_20260101.csv", fmt="csv")   # 给交易终端
export_signals(sig, "signals_20260101.json", fmt="json") # 给程序化接口

输出即一份可直接落盘的信号文件——研究到此闭环。

6. 坑与注意事项

  1. 信号滞后:持仓信号要用 t-1 的因子算 t 的持仓,避免偷看当天数据。
  2. 复权一致:因子与回测的价格必须同口径(都前复权或都后复权),否则收益对不上。
  3. 停牌/退市:信号里的股票若停牌,实盘下不了单,导出时过滤掉当日无交易的代码。
  4. 文件编码:CSV 用 utf-8-sig(带 BOM),Excel 打开中文列名不乱码;JSON 用 ensure_ascii=False。
  5. 增量更新:每天只导出当天新增信号,别全量覆盖历史,便于复盘与归因。

7. 小结与下一篇预告

本篇把前面 31 篇串成一条「取数 → 因子 → 信号 → 回测 → 导出」的可落盘链路,generate_signals / export_signals / performance_report 三件套直接复用全系列结论。至此《Python量化实战全攻略》32 篇全部完结——从数据底座到实盘信号,一站打通。后续若想扩展,可把 #31 的参数优化接进本篇的 pipeline,做自动调参 + 信号再导出。

下一篇预告:系列已完结(全 32 篇收官)。

8. 免责声明

本文所有接口路径均来自官方文档,示例数据为演示用途,不构成任何投资建议。实盘使用前请自行核对因子口径、复权与停牌处理。


关于数据来源:本文接口来自智兔数服(www.zhituapi.com)官方文档,覆盖沪深A股、指数、基金、可转债等 200+ 已验证接口路径,注册可领免费 token 做研究测试。
更多实战手册见技术博客:zhituapi.com/blog.html。
《Python量化实战全攻略》共 32 篇,至此全系列收官——数据底座、因子、择时、组合风控、参数优化、信号导出一站打通。

想亲自试一下?免费获取证书