Python实现ETF量化策略:横截面动量轮动 + 收益率排序 + Top-K持仓

"""================================================================================动量策略（美股ETF版）：横截面动量轮动- 只做多，每日调仓- 计算过去 N 日收益率，选择排名前 K 只 ETF 等权持有- 数据来源：本地 CSV 文件（列名：date,open,high,low,close,volume）================================================================================"""import pandas as pdimport numpy as npfrom datetime import datetimeimport matplotlibmatplotlib.use('TkAgg')import matplotlib.pyplot as pltfrom dataclasses import dataclassfrom typing import Dict, List, Optional, Tupleimport warningsimport oswarnings.filterwarnings('ignore')plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']plt.rcParams['axes.unicode_minus'] = False@dataclassclass TradeRecord:    date: datetime    code: str    action: str    shares: int    price: float    value: float    pnl: float = 0.0    reason: str = ""class MomentumStrategy:    def __init__(            self,            start_date: str = "20190101",            end_date: str = "20260410",            lookback: int = 60,          # 动量计算窗口（交易日）            top_k: int = 3,              # 选择表现最强的 K 只            initial_capital: float = 1_000_000,            commission: float = 0.0001,            slippage: float = 0.0002,            local_data_path: str = r"你的数据路径",    ):        self.start_date = start_date        self.end_date = end_date        self.lookback = lookback        self.top_k = top_k        self.initial_capital = initial_capital        self.commission = commission        self.slippage = slippage        self.local_data_path = local_data_path        # 美股 ETF 标的池（11只，与均值回归策略相同）        self.etf_pool = [            "XLE", "XLF", "XLK", "XLU", "XLV",            "XLY", "XLP", "XLI", "XLB", "XLRE", "XLC"        ]        self.close_data: Dict[str, pd.Series] = {}        self.open_data: Dict[str, pd.Series] = {}        self.trades: List[TradeRecord] = []        self.positions: Dict[str, Dict] = {}        self.daily_stats = []        self.current_capital = initial_capital        self.peak_capital = initial_capital        self.trading_dates = []    # -------------------- 数据读取 --------------------    def _read_local_etf_file(self, code: str) -> Optional[Tuple[pd.Series, pd.Series]]:        file_path = os.path.join(self.local_data_path, f"{code}.csv")        if not os.path.exists(file_path):            print(f"  ✗ {code}: 文件不存在")            return None        try:            df = pd.read_csv(file_path, encoding='utf-8')        except UnicodeDecodeError:            df = pd.read_csv(file_path, encoding='gbk')        df['date'] = pd.to_datetime(df['date'])        df.set_index('date', inplace=True)        close = df['close'].astype(float)        open_ = df['open'].astype(float)        return close, open_    def fetch_data(self):        print("\n[数据加载]")        for code in self.etf_pool:            result = self._read_local_etf_file(code)            if result:                close, open_ = result                self.close_data[code] = close                self.open_data[code] = open_        all_dates = [set(d.index) for d in self.close_data.values()]        self.trading_dates = sorted(set.intersection(*all_dates))        self.start_idx = self.lookback + 1        print(f"共同交易日: {len(self.trading_dates)} 天")    # -------------------- 动量信号生成（横截面）--------------------    def generate_signals(self, current_date: datetime) -> Dict[str, str]:        momentum_scores = {}        for code in self.close_data:            hist = self.close_data[code][self.close_data[code].index <= current_date].tail(self.lookback)            if len(hist) < self.lookback:                momentum_scores[code] = -np.inf                continue            ret = (hist.iloc[-1] - hist.iloc[0]) / hist.iloc[0]            momentum_scores[code] = ret        sorted_codes = sorted(momentum_scores.keys(), key=lambda x: momentum_scores[x], reverse=True)        selected = [c for c in sorted_codes if momentum_scores[c] > -np.inf][:self.top_k]        signals = {}        for code in list(self.positions.keys()):            if code not in selected:                signals[code] = 'SELL'        for code in selected:            if code not in self.positions:                signals[code] = 'BUY'        return signals    # -------------------- 交易执行 --------------------    def execute_trades(self, date: datetime, signals: Dict[str, str], next_opens: Dict[str, float]):        # 先平仓        for code, action in signals.items():            if action == 'SELL' and code in self.positions:                pos = self.positions[code]                price = next_opens.get(code)                if price is None:                    continue                gross = pos['shares'] * price                fee = gross * (self.commission + self.slippage)                net = gross - fee                self.current_capital += net                pnl = net - (pos['shares'] * pos['entry_price'])                self.trades.append(TradeRecord(date, code, 'SELL', pos['shares'], price, net, pnl, '动量轮动卖出'))                del self.positions[code]        # 再开仓        buy_codes = [c for c, a in signals.items() if a == 'BUY']        if buy_codes:            capital_per = self.current_capital / len(buy_codes)            for code in buy_codes:                price = next_opens.get(code)                if price is None:                    continue                shares = int(capital_per / price)                if shares < 1:                    continue                cost = shares * price * (1 + self.commission + self.slippage)                if cost > self.current_capital:                    shares = int(self.current_capital / (price * (1 + self.commission + self.slippage)))                    if shares < 1:                        continue                    cost = shares * price * (1 + self.commission + self.slippage)                self.current_capital -= cost                self.positions[code] = {'shares': shares, 'entry_price': price}                self.trades.append(TradeRecord(date, code, 'BUY', shares, price, -cost, 0, '动量轮动买入'))    # -------------------- 回测引擎 --------------------    def run_backtest(self):        print("\n[回测运行]")        for i in range(self.start_idx, len(self.trading_dates) - 1):            curr = self.trading_dates[i]            nxt = self.trading_dates[i+1]            signals = self.generate_signals(curr)            next_opens = {c: self.open_data[c].loc[nxt] for c in self.close_data if nxt in self.open_data[c].index}            self.execute_trades(curr, signals, next_opens)            total = self.current_capital            for code, pos in self.positions.items():                if curr in self.close_data[code].index:                    total += pos['shares'] * self.close_data[code].loc[curr]            self.peak_capital = max(self.peak_capital, total)            dd = (self.peak_capital - total) / self.peak_capital if self.peak_capital > 0 else 0            self.daily_stats.append({'date': curr, 'total': total, 'dd': dd})        return pd.DataFrame(self.daily_stats)    # -------------------- 绩效报告 --------------------    def report(self, stats_df):        # ...（与均值回归策略相同，此处省略以节省篇幅，完整代码见文末）        pass    def run(self):        self.fetch_data()        stats = self.run_backtest()        self.report(stats)if __name__ == "__main__":    strategy = MomentumStrategy(        local_data_path=r"你的数据路径",        start_date="20190101",        end_date="20260410",        lookback=60,        top_k=3,        initial_capital=1_000_000,        commission=0.0001,        slippage=0.0002,    )    strategy.run()