"""quantlab 命令行入口。
用法示例：
python run.py --risk 稳健 --horizon 波段
python run.py --risk 进取 --horizon 中线 --style momentum=0.4,low_vol=0.2,trend=0.4 --positions 8
python run.py --preset configs/稳健-波段.json --report out/report.html
"""
from __future__ import annotations
import argparse
import json
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from quantlab import dataloader as dl
from quantlab.backtest import Backtester
from quantlab.metrics import monthly_returns, summarize, yearly_table
from quantlab.report import build_report
from quantlab.periods import PERIOD_DAYS, PERIOD_PROFILE
from quantlab.strategy_generator import explain, generate_strategy
ROOT = os.path.dirname(os.path.abspath(__file__))
QLAB_DATA = os.path.join(ROOT, "quantlab", "data")
DATA_START = "2021-06-01" # 预留 250 日因子预热窗口
def parse_style(s: str):
if not s:
return None
out = {}
for kv in s.split(","):
k, v = kv.split("=")
out[k.strip()] = float(v)
return out
def load_universe_codes(universe: str):
if universe in ("sample", "hs300_sample"):
path = os.path.join(QLAB_DATA, "universe_sample.csv")
else:
path = universe if os.path.exists(universe) else os.path.join(QLAB_DATA, f"universe_{universe}.csv")
if not os.path.exists(path):
raise FileNotFoundError(f"股票池文件不存在: {path}")
codes = []
with open(path, encoding="utf-8-sig") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#") or line.startswith("code,"):
continue
codes.append(line.split(",")[0])
return codes
def main():
ap = argparse.ArgumentParser(description="偏好参数驱动的量化策略生成与回测")
ap.add_argument("--risk", default="平衡", help="保守/稳健/平衡/进取/激进")
ap.add_argument("--horizon", default="波段", help="短线/波段/中线/长线，或周期别名 周/双周/月/季/半年/年")
ap.add_argument("--style", default="", help="因子权重覆盖，如 momentum=0.4,low_vol=0.3,trend=0.3")
ap.add_argument("--positions", type=int, default=None)
ap.add_argument("--rebalance", type=int, default=None, help="调仓间隔（交易日）")
ap.add_argument("--cash", type=float, default=None, help="现金缓冲比例")
ap.add_argument("--start", default="2022-01-01")
ap.add_argument("--end", default="2030-01-01")
ap.add_argument("--universe", default="sample")
ap.add_argument("--limit", type=int, default=None, help="最多使用多少只标的（调试用）")
ap.add_argument("--preset", default=None, help="从 JSON 文件加载完整策略配置，跳过偏好推导")
ap.add_argument("--report", default=None, help="HTML 报告输出路径")
ap.add_argument("--save-strategy", default=None)
ap.add_argument("--no-trend-filter", action="store_true")
args = ap.parse_args()
# 周期别名（周/双周/月/季/半年/年）映射到内部 horizon
horizon = args.horizon
if horizon in PERIOD_DAYS:
horizon = PERIOD_PROFILE[PERIOD_DAYS[horizon]]["horizon"]
if args.preset:
with open(args.preset, encoding="utf-8") as f:
strategy = json.load(f)
else:
strategy = generate_strategy(
risk=args.risk, horizon=horizon, style=parse_style(args.style),
max_positions=args.positions, rebalance_days=args.rebalance,
cash_buffer=args.cash, universe=args.universe,
start=args.start, end=args.end,
extra_risk={"trend_filter": not args.no_trend_filter},
)
print("\n".join(explain(strategy)))
print()
codes = load_universe_codes(args.universe)
if args.limit:
codes = codes[: args.limit]
print(f"股票池 {len(codes)} 只，加载日线数据（{DATA_START} 起，含预热窗口）...")
data_map = dl.load_universe(codes, DATA_START, args.end)
print(f"有效标的 {len(data_map)} 只")
bench = dl.fetch_index(strategy["universe"]["benchmark"], DATA_START, args.end)
bench_close = bench["close"] if len(bench) else None
print("回测运行中...")
bt = Backtester(strategy)
res = bt.run(data_map, bench_close)
rets = res["strat_rets"]
curve = res["curve"]["equity"]
eq_for_metrics = (1 + rets).cumprod()
bench_rets = None
if len(res["benchmark"]):
b = res["benchmark"]["equity"]
bench_rets = b.pct_change().dropna()
bench_rets = bench_rets[rets.index.intersection(bench_rets.index)]
m = summarize(rets, eq_for_metrics, bench_rets,
turnover_hist=res["turnover_hist"], ic_series=res["ic_series"])
yr = yearly_table(rets)
names = {}
upath = os.path.join(QLAB_DATA, "universe_sample.csv")
if os.path.exists(upath):
with open(upath, encoding="utf-8-sig") as f:
for line in f:
if "," in line and not line.startswith("#") and not line.startswith("code,"):
c, n = line.strip().split(",", 1)
names[c] = n
print("\n===== 回测绩效 =====")
for k, v in m.items():
if k in ("total_return", "cagr", "ann_vol", "max_drawdown", "daily_win_rate",
"excess_cagr_vs_bench", "tracking_error"):
print(f" {k:24s} {v:>10.2%}")
elif k in ("n_days", "n_ic_obs"):
print(f" {k:24s} {v:>10d}")
else:
print(f" {k:24s} {v:>10.3f}")
print("\n===== 分年度收益 =====")
print(yr.to_string(float_format=lambda x: f"{x:.2%}"))
# 保存产物
outdir = os.path.join(ROOT, "out")
os.makedirs(outdir, exist_ok=True)
tag = f"{strategy['preferences']['risk']}_{strategy['preferences']['horizon']}"
strat_path = args.save_strategy or os.path.join(outdir, f"strategy_{tag}.json")
with open(strat_path, "w", encoding="utf-8") as f:
json.dump(strategy, f, ensure_ascii=False, indent=2)
sources = sorted({str(df.attrs.get("source", "unknown")) for df in data_map.values()})
result_payload = {
"strategy": strategy,
"metrics": m,
"curve": [{"date": str(d.date()), "equity": round(float(v), 6)}
for d, v in curve.items()],
"benchmark": [{"date": str(d.date()), "equity": round(float(v), 6)}
for d, v in res["benchmark"]["equity"].items()] if len(res["benchmark"]) else [],
"monthly": [{"month": str(idx)[:7], "ret": round(float(v), 6)} for idx, v in monthly_returns(rets).items()],
"yearly": {str(k.year): round(float(v), 6) for k, v in yr.items()},
"exposure": res["exposure_hist"],
"ic": res["ic_series"],
"latest_holdings": [
{"code": c, "name": names.get(c, ""), "weight": round(w, 4)}
for c, w in sorted(res["latest_weights"].items(), key=lambda x: -x[1])
],
"last_rebalance": res["rebalance_log"][-1] if res["rebalance_log"] else None,
"data_notes": {
"kline_source": "日线来源: " + "、".join(sources) + "（tencent_qfq=腾讯前复权；sina_unadjusted=新浪不复权，除权日存在价格跳空会轻微影响动量类因子）",
"universe": f"内置大市值样本池 {len(data_map)} 只有效标的（静态列表，存在幸存者偏差，成分快照见 data/universe_sample.csv）",
"benchmark": "沪深300（腾讯接口 sh000300）",
"cost_assumption": strategy["execution"],
},
}
result_path = os.path.join(outdir, f"result_{tag}.json")
with open(result_path, "w", encoding="utf-8") as f:
json.dump(result_payload, f, ensure_ascii=False, indent=2)
if args.report:
html = build_report(result_payload)
with open(args.report, "w", encoding="utf-8") as f:
f.write(html)
print(f"\nHTML 报告: {args.report}")
print(f"策略规格: {strat_path}\n回测明细: {result_path}")
if __name__ == "__main__":
main()