【Python量化实战全攻略 #32】一键回测与实盘信号导出手册:从研究到落盘(附代码)
摘要:【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. 坑与注意事项
- 信号滞后:持仓信号要用
t-1的因子算t的持仓,避免偷看当天数据。 - 复权一致:因子与回测的价格必须同口径(都前复权或都后复权),否则收益对不上。
- 停牌/退市:信号里的股票若停牌,实盘下不了单,导出时过滤掉当日无交易的代码。
- 文件编码:CSV 用
utf-8-sig(带 BOM),Excel 打开中文列名不乱码;JSON 用ensure_ascii=False。 - 增量更新:每天只导出当天新增信号,别全量覆盖历史,便于复盘与归因。
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 篇,至此全系列收官——数据底座、因子、择时、组合风控、参数优化、信号导出一站打通。