Python实现ETF量化策略:时间序列趋势跟踪

"""================================================================================趋势跟踪策略（美股ETF版）：双均线金叉/死叉- 只做多，每日调仓（每只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 TrendFollowingStrategy:    def __init__(            self,            start_date: str = "20190101",            end_date: str = "20260410",            # 双均线参数            short_window: int = 10,            long_window: int = 30,            # 资金与风控            initial_capital: float = 1_000_000,            commission: float = 0.0001,            slippage: float = 0.0002,            local_data_path: str = r"你的数据路径",            # 仓位管理：每只ETF分配资金的百分比            capital_per_etf_pct: float = 0.10,    ):        self.start_date = start_date        self.end_date = end_date        self.short_window = short_window        self.long_window = long_window        self.initial_capital = initial_capital        self.commission = commission        self.slippage = slippage        self.local_data_path = local_data_path        self.capital_per_etf_pct = capital_per_etf_pct        # 美股 ETF 标的池（与前两篇相同）        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.long_window + 1        print(f"共同交易日: {len(self.trading_dates)} 天")    # -------------------- 双均线信号生成 --------------------    def generate_signals(self, current_date: datetime) -> Dict[str, str]:        signals = {}        for code in self.close_data:            price_series = self.close_data[code]            hist = price_series[price_series.index <= current_date]            if len(hist) < self.long_window + 1:                continue            ma_short = hist.rolling(window=self.short_window).mean()            ma_long = hist.rolling(window=self.long_window).mean()            curr_short = ma_short.iloc[-1]            curr_long = ma_long.iloc[-1]            prev_short = ma_short.iloc[-2] if len(ma_short) >= 2 else np.nan            prev_long = ma_long.iloc[-2] if len(ma_long) >= 2 else np.nan            if pd.isna(curr_short) or pd.isna(curr_long) or pd.isna(prev_short) or pd.isna(prev_long):                continue            is_holding = code in self.positions            # 金叉：短期上穿长期            if not is_holding and (prev_short <= prev_long) and (curr_short > curr_long):                signals[code] = 'BUY'            # 死叉：短期下穿长期            elif is_holding and (prev_short >= prev_long) and (curr_short < curr_long):                signals[code] = 'SELL'        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']        for code in buy_codes:            price = next_opens.get(code)            if price is None:                continue            risk_capital = self.current_capital * self.capital_per_etf_pct            shares = int(risk_capital / 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):        if stats_df.empty:            print("无回测数据")            return        final = stats_df.iloc[-1]['total']        ret = (final - self.initial_capital) / self.initial_capital * 100        days = len(stats_df)        years = days / 252        ann_ret = (final / self.initial_capital) ** (1/years) - 1 if years > 0 else 0        stats_df['ret'] = stats_df['total'].pct_change()        vol = stats_df['ret'].std() * np.sqrt(252) * 100        sharpe = (ann_ret*100 - 3) / vol if vol > 0 else 0        max_dd = stats_df['dd'].max() * 100        closed = [t for t in self.trades if t.action == 'SELL']        wins = [t for t in closed if t.pnl > 0]        win_rate = len(wins)/len(closed)*100 if closed else 0        print("\n" + "="*60)        print("趋势跟踪策略（双均线金叉/死叉）- 美股ETF版")        print("="*60)        print(f"回测区间: {self.start_date} ~ {self.end_date} ({days}天)")        print(f"初始资金: {self.initial_capital:,.0f}  期末: {final:,.0f}")        print(f"总收益率: {ret:+.2f}%   年化: {ann_ret*100:+.2f}%")        print(f"年化波动: {vol:.2f}%   夏普: {sharpe:.2f}")        print(f"最大回撤: {max_dd:.2f}%")        print(f"交易次数: {len(closed)} 笔   胜率: {win_rate:.1f}%")        if closed:            avg_pnl = np.mean([t.pnl for t in closed])            print(f"平均盈亏: {avg_pnl:,.0f}")        print("="*60)        # 绘图        fig, ax = plt.subplots(2, 1, figsize=(12, 8))        ax[0].plot(stats_df['date'], stats_df['total'], label='NAV', color='blue')        ax[0].axhline(self.initial_capital, color='gray', linestyle='--')        ax[0].set_title('Trend Following Strategy (Dual Moving Average) - US ETFs')        ax[0].legend()        ax[1].fill_between(stats_df['date'], 0, stats_df['dd']*100, color='red', alpha=0.3)        ax[1].set_ylabel('Drawdown %')        plt.tight_layout()        plt.show()    def run(self):        self.fetch_data()        stats = self.run_backtest()        self.report(stats)        if self.trades:            pd.DataFrame([t.__dict__ for t in self.trades]).to_csv('trades_trend_following_us.csv', index=False, encoding='utf-8-sig')            print("交易记录已保存至 trades_trend_following_us.csv")if __name__ == "__main__":    strategy = TrendFollowingStrategy(        local_data_path=r"你的数据路径",        start_date="20190101",        end_date="20260410",        short_window=5,        long_window=10,        initial_capital=1_000_000,        commission=0.0001,        slippage=0.0002,        capital_per_etf_pct=0.10,    )    strategy.run()