参数扫描结果：lookback=5, top_n=1 最优，收益 94.01%
核心逻辑：每周一买入过去 5 天涨幅最大的 1 只 ETF，全仓持有，次周一轮动。
代码部分
# -*- coding: utf-8 -*- """动量策略：每周一买入过去 5 天涨幅最大的 1 只 ETF，次周轮动""" from value300 import Value300 client = Value300() def backtest(etfs, lookback=5, top_n=1, capital=100000): all_data = {e: client.get_history(e, days=365, fields="minimal") for e in etfs} dates = [d["trade_date"] for d in list(all_data.values())[0]] import datetime mondays = [d for d in dates if datetime.datetime.strptime(d, "%Y-%m-%d").weekday() == 0] positions = {} trades = [] for date in mondays: # 动量排名 ranks = [] for code, data in all_data.items(): try: idx = next(i for i, d in enumerate(data) if d["trade_date"] == date) except StopIteration: continue if idx < lookback: continue mom = (data[idx]["close"] / data[idx - lookback]["close"] - 1) * 100 ranks.append({"code": code, "mom": mom, "price": data[idx]["close"]}) ranks.sort(key=lambda x: x["mom"], reverse=True) selected = ranks[:top_n] # 卖出 for code, shares in list(positions.items()): data = all_data[code] try: idx = next(i for i, d in enumerate(data) if d["trade_date"] == date) capital += shares * data[idx]["close"] except (StopIteration, IndexError): pass positions = {} # 买入 if selected and capital > 0: per = capital / len(selected) for s in selected: shares = int(per / s["price"]) if shares: positions[s["code"]] = shares capital -= shares * s["price"] value = capital + sum( shares * next(d["close"] for d in all_data[code] if d["trade_date"] == date) for code, shares in positions.items() ) trades.append({"date": date, "value": value}) ret = (trades[-1]["value"] / 100000 - 1) * 100 print(f"初始: 100,000 → 终值: {trades[-1]['value']:,.0f} 收益: {ret:.2f}%") return trades if __name__ == "__main__": trades = backtest(["510300", "510500", "159915", "588000", "510880"]) for t in trades: print(f"{t['date']} {t['value']:>10,.0f}")
)
