"""下个周期推荐（CLI）。
与回测的区别：回测回答"这套规则历史上表现如何"，本工具回答
"按这套因子与周期，**下个周期该买哪几只、各配多少**"。
用法：
python recommend.py --period 月
python recommend.py --period 周 --style momentum=0.3,reversal=0.3,trend=0.4 --top 10
python recommend.py --days 15 --risk 进取 --dashboard out/dashboard.html
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import numpy as np
import pandas as pd
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from quantlab import dataloader as dl
from quantlab.factors import cross_section_z, FACTOR_FUNCS
from quantlab.periods import PERIOD_PROFILE, PERIOD_DAYS, resolve_period
from quantlab.strategy_generator import DEFAULT_STYLE, RISK_LEVELS, normalize_style
ROOT = os.path.dirname(os.path.abspath(__file__))
QLAB_DATA = os.path.join(ROOT, "quantlab", "data")
DATA_START = "2021-06-01"
# 需要按周期切换回看窗口的因子
PERIODIC_FACTORS = {"momentum": "mom", "trend": "trend"}
# 固定回看窗口的因子
FIXED_LOOKBACK = {"quality_momentum": 120, "reversal": 5, "low_vol": 60, "high_52w": 250, "liquidity": 20}
# 快照中需要预计算的动量/趋势窗口（覆盖全部周期档位）
MOM_WINDOWS = [20, 60, 120, 180, 250]
TREND_WINDOWS = [20, 60, 120, 200, 250]
FACTOR_LABELS = {
"momentum": "动量",
"quality_momentum": "质量动量",
"low_vol": "低波",
"trend": "趋势",
"reversal": "短期反转",
"high_52w": "52周高点",
"liquidity": "流动性",
}
def load_names() -> dict:
names = {}
path = os.path.join(QLAB_DATA, "universe_sample.csv")
if os.path.exists(path):
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
parts = line.split(",", 1)
if len(parts) == 2:
names[parts[0]] = parts[1]
return names
def load_codes() -> list:
codes = []
path = os.path.join(QLAB_DATA, "universe_sample.csv")
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 build_snapshot(data_map: dict, asof: str = None) -> dict:
"""在 asof 日（默认全池最新共同交易日）计算每只股票的因子值快照。"""
# 选择 asof：优先用出现频率最高的最新日期
last_dates = [df.index.max() for df in data_map.values()]
cutoff = max(last_dates) if not asof else pd.Timestamp(asof)
counts = {}
for df in data_map.values():
if cutoff in df.index:
counts[cutoff] = counts.get(cutoff, 0) + 1
else:
prev = df.index[df.index <= cutoff]
if len(prev):
d = prev[-1]
counts[d] = counts.get(d, 0) + 1
if not counts:
raise ValueError("无可用交易日数据")
# 取"最近且覆盖标的数达到最大值 90%"的日期
max_cov = max(counts.values())
candidates = sorted([d for d, c in counts.items() if c >= max_cov * 0.9])
asof_date = candidates[-1]
if asof and pd.Timestamp(asof) < asof_date:
asof_date = pd.Timestamp(asof)
stocks = []
for code, df in data_map.items():
sub = df.loc[:asof_date]
if len(sub) < 260:
continue
close = sub["close"]
ret = close.pct_change()
ma200 = close.rolling(200, min_periods=150).mean().iloc[-1]
row = {
"code": code,
"close": round(float(close.iloc[-1]), 2),
"ma200_dev": round(float(close.iloc[-1] / ma200 - 1.0), 4) if ma200 == ma200 else None,
"vol60": round(float(ret.tail(60).std() * np.sqrt(252)), 4),
}
for w in MOM_WINDOWS:
key = f"mom{w}"
if len(close) > w + 5:
row[key] = round(float(close.iloc[-6] / close.iloc[-6 - w] - 1.0), 4)
else:
row[key] = None
for w in TREND_WINDOWS:
key = f"trend{w}"
if len(close) > w:
ma = close.rolling(w, min_periods=max(10, w // 3)).mean().iloc[-1]
row[key] = round(float(close.iloc[-1] / ma - 1.0), 4) if ma == ma else None
else:
row[key] = None
# 固定窗口因子
for fname, w in FIXED_LOOKBACK.items():
fn, _ = FACTOR_FUNCS[fname]
try:
s = fn(sub, w).dropna()
row[fname] = round(float(s.iloc[-1]), 6) if len(s) else None
except Exception:
row[fname] = None
stocks.append(row)
return {"as_of": str(asof_date.date()), "stocks": stocks}
def _pick_key(factor: str, period_cfg: dict) -> str:
if factor == "momentum":
return f"mom{period_cfg['momentum']}"
if factor == "trend":
return f"trend{period_cfg['trend']}"
return factor
def score_snapshot(snapshot: dict, weights: dict, period_cfg: dict, top_n: int = 12,
trend_filter: bool = True, cash_buffer: float = 0.05,
max_weight: float = 0.12) -> dict:
"""按因子权重给快照打分并选出 TopN，返回打分明细与目标权重。"""
stocks = snapshot["stocks"]
if not stocks:
raise ValueError("快照为空")
z_by_factor = {}
for f, w in weights.items():
if w <= 0:
continue
key = _pick_key(f, period_cfg)
series = pd.Series({st["code"]: st.get(key) for st in stocks}, dtype="float64")
valid = series.dropna()
if len(valid) < 8:
continue
z = cross_section_z(valid.to_frame("v").T).iloc[0]
z_by_factor[f] = z
rows = []
for st in stocks:
code = st["code"]
# 趋势过滤：价格低于 200 日线不入选
if trend_filter and (st.get("ma200_dev") is None or st["ma200_dev"] < 0):
continue
total, contrib = 0.0, {}
covered = 0
for f, z in z_by_factor.items():
v = z.get(code, np.nan)
if v == v:
c = weights[f] * float(v)
contrib[f] = round(c, 4)
total += c
covered += 1
if covered < max(2, len(z_by_factor) // 2):
continue
rows.append({"code": code, "score": total, "contrib": contrib, **{
k: st.get(k) for k in ("close", "ma200_dev", "vol60")
}})
rows.sort(key=lambda r: -r["score"])
selected = rows[:top_n]
base = 1.0 - cash_buffer
w_each = min(max_weight, base / len(selected)) if selected else 0.0
for r in rows:
r["weight"] = 0.0
for r in selected:
r["weight"] = round(w_each, 4)
return {
"as_of": snapshot["as_of"],
"period": period_cfg,
"weights": weights,
"ranked": rows,
"selected": selected,
"invested": round(w_each * len(selected), 4),
"n_candidates": len(rows),
}
def print_result(res: dict, names: dict):
p = res["period"]
print(f"\n交易日 {res['as_of']} ｜ 推荐周期：{p['label']}（{p['days']} 个交易日）"
f" ｜ 动量窗口 {p['momentum']} 日 ｜ 趋势窗口 {p['trend']} 日")
print("因子权重：" + "，".join(
f"{FACTOR_LABELS.get(k, k)} {v:.0%}" for k, v in sorted(res["weights"].items(), key=lambda x: -x[1]) if v > 0))
print(f"候选 {res['n_candidates']} 只（已过 200 日线趋势过滤），推荐 {len(res['selected'])} 只，"
f"合计仓位 {res['invested']:.1%}\n")
header = f"{'#':>3} {'代码':<8}{'名称':<10}{'得分':>8}{'权重':>8}{'收盘':>9}{'距MA200':>9}{'年化波动':>9}"
print(header)
print("-" * len(header) + "-" * 6)
for i, r in enumerate(res["selected"], 1):
print(f"{i:>3} {r['code']:<8}{names.get(r['code'], ''):<10}{r['score']:>8.3f}{r['weight']:>8.1%}"
f"{r['close']:>9.2f}{(r['ma200_dev'] or 0):>8.1%}{(r['vol60'] or 0):>8.1%}")
print("\n各票因子贡献（得分 = Σ 权重×因子Z值）：")
for i, r in enumerate(res["selected"][:8], 1):
detail = "，".join(f"{FACTOR_LABELS.get(k, k)}{v:+.2f}" for k, v in
sorted(r["contrib"].items(), key=lambda x: -abs(x[1]))[:4])
print(f" {i}. {r['code']} {names.get(r['code'], ''):<6} {detail}")
def main():
ap = argparse.ArgumentParser(description="下个周期推荐股票（因子+周期可调）")
ap.add_argument("--period", default="月", help="周/双周/月/季/半年/年")
ap.add_argument("--days", type=int, default=None, help="自定义周期（交易日数，3~250），优先于 --period")
ap.add_argument("--style", default="", help="因子权重，如 momentum=0.4,low_vol=0.3,trend=0.3；可用 0 关闭某因子")
ap.add_argument("--risk", default="平衡", help="保守/稳健/平衡/进取/激进（决定持仓数与单票上限）")
ap.add_argument("--top", type=int, default=None, help="推荐股票数量")
ap.add_argument("--asof", default=None, help="指定信号日 YYYY-MM-DD（默认最新交易日）")
ap.add_argument("--no-trend-filter", action="store_true", help="关闭 200 日线趋势过滤")
ap.add_argument("--refresh", action="store_true", help="强制刷新行情缓存")
ap.add_argument("--json", default=None, help="推荐结果 JSON 输出路径")
ap.add_argument("--dashboard", default=None, help="生成交互式选股看板 HTML")
args = ap.parse_args()
period_cfg = resolve_period(args.period, args.days)
risk = RISK_LEVELS[args.risk]
# 默认权重：默认风格按周期调整反转权重
weights = dict(DEFAULT_STYLE)
weights["reversal"] = period_cfg["reversal"]
if args.style:
override = {}
for kv in args.style.split(","):
k, v = kv.split("=")
override[k.strip()] = float(v)
weights.update(override)
weights = normalize_style(weights)
top_n = args.top or risk["max_positions"]
codes = load_codes()
print(f"加载 {len(codes)} 只标的行情{'（强制刷新）' if args.refresh else ''}...")
data_map = dl.load_universe(codes, DATA_START, "2030-01-01", use_cache=not args.refresh)
print(f"有效标的 {len(data_map)} 只")
snapshot = build_snapshot(data_map)
res = score_snapshot(snapshot, weights, period_cfg, top_n=top_n,
trend_filter=not args.no_trend_filter,
cash_buffer=risk["cash_buffer"], max_weight=risk["max_weight"])
names = load_names()
print_result(res, names)
# 多周期对比：同一套因子下，不同周期各自推荐谁
print("\n各周期推荐对比（Top5）：")
for label, days in PERIOD_DAYS.items():
pc = resolve_period(label)
r = score_snapshot(snapshot, weights, pc, top_n=5,
trend_filter=not args.no_trend_filter,
cash_buffer=risk["cash_buffer"], max_weight=risk["max_weight"])
picks = "、".join(f"{x['code']}{names.get(x['code'], '')}" for x in r["selected"])
print(f" {label:<3}({days:>3}日) {picks}")
os.makedirs(os.path.join(ROOT, "out"), exist_ok=True)
payload = {
"as_of": res["as_of"],
"period": res["period"],
"weights": res["weights"],
"factor_labels": FACTOR_LABELS,
"invested": res["invested"],
"n_candidates": res["n_candidates"],
"recommendations": [
{"rank": i, "code": r["code"], "name": names.get(r["code"], ""),
"score": round(r["score"], 4), "weight": r["weight"], "close": r["close"],
"ma200_dev": r["ma200_dev"], "vol60": r["vol60"], "contrib": r["contrib"]}
for i, r in enumerate(res["selected"], 1)
],
"data_notes": {
"kline_source": "腾讯/新浪日线（公开接口）",
"universe": f"内置大市值样本池（{len(data_map)} 只有效标的，静态列表含幸存者偏差）",
"method": "截面 MAD 去极值 + Z-score 加权合成，200 日线之上才入选",
},
}
out_json = args.json or os.path.join(ROOT, "out", f"recommend_{period_cfg['label']}.json")
with open(out_json, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, indent=2)
print(f"\n推荐清单 JSON: {out_json}")
if args.dashboard:
from quantlab.dashboard import build_dashboard
html = build_dashboard(snapshot, names, period_cfg, weights, top_n)
with open(args.dashboard, "w", encoding="utf-8") as f:
f.write(html)
print(f"交互看板: {args.dashboard}")
if __name__ == "__main__":
main()