CTA趋势跟踪策略：从原理到Python实战（附完整代码）

import numpy as np
import pandas as pd
from typing import Dict, List, Tuple


class TrendFollowingSystem:
    """
    多品种趋势跟踪系统
    """

    def __init__(
        self,
        lookback_months: List[int] = [1, 3, 6, 12],
        vol_target: float = 0.20,
        vol_lookback: int = 60,
        max_leverage: float = 2.0,
    ):
        self.lookback_months = lookback_months
        self.vol_target = vol_target
        self.vol_lookback = vol_lookback
        self.max_leverage = max_leverage

    def compute_signals(self, prices: Dict[str, pd.DataFrame]) -> Dict[str, pd.Series]:
        """
        计算趋势信号
        prices: {品种名: DataFrame(index=date, columns=['close','high','low','volume'])}
        返回: {品种名: Series(index=date) 信号值}
        """
        signals = {}

        for asset, df in prices.items():
            close = df['close']

            # 计算每个回看期的动量信号
            momentum_signals = []
            for m in self.lookback_months:
                # 假设每月21个交易日
                days = m * 21
                # 指数加权收益
                returns = close.pct_change(days)
                momentum_signals.append(returns)

            # 多信号等权融合
            raw_signal = pd.concat(momentum_signals, axis=1).mean(axis=1)

            # 波动率标准化
            daily_returns = close.pct_change()
            realized_vol = (
                daily_returns.rolling(self.vol_lookback).std() * np.sqrt(252)
            )
            normalized_signal = raw_signal / realized_vol  # 或使用 tanh 压缩
            normalized_signal = np.tanh(normalized_signal)  # 限制极端值

            signals[asset] = normalized_signal

        return signals

    def compute_positions(
        self, signals: Dict[str, pd.Series]
    ) -> Dict[str, pd.Series]:
        """
        根据信号和波动率目标计算头寸
        """
        positions = {}

        for asset, signal in signals.items():
            # 头寸 = 目标波动率 / 品种数量 * 信号
            n_assets = len(signals)
            pos = (self.vol_target / n_assets) * signal.fillna(0)

            # 限制单品种杠杆
            pos = pos.clip(-self.max_leverage / n_assets, self.max_leverage / n_assets)
            positions[asset] = pos

        return positions

    def backtest(
        self, prices: Dict[str, pd.DataFrame], trading_cost_bps: float = 5.0
    ) -> pd.DataFrame:
        """
        回测
        """
        signals = self.compute_signals(prices)
        positions = self.compute_positions(signals)

        # 组合每日收益
        portfolio_returns = pd.DataFrame(index=prices[list(prices.keys())[0]].index)

        for asset, df in prices.items():
            daily_returns = df['close'].pct_change()
            # 使用前一日头寸*今日收益（避免前视偏差）
            strategy_returns = positions[asset].shift(1) * daily_returns
            portfolio_returns[asset] = strategy_returns

        portfolio_returns['total'] = portfolio_returns.sum(axis=1)

        # 考虑交易成本
        if trading_cost_bps > 0:
            for asset in prices:
                turnover = positions[asset].diff().abs()
                cost = turnover * trading_cost_bps / 10000
                portfolio_returns[asset] -= cost

        portfolio_returns['total'] = portfolio_returns[
            [c for c in portfolio_returns.columns if c != 'total']
        ].sum(axis=1)

        return portfolio_returns

    def performance_stats(self, returns: pd.Series) -> Dict:
        """计算绩效指标"""
        ann_return = returns.mean() * 252
        ann_vol = returns.std() * np.sqrt(252)
        sharpe = ann_return / ann_vol if ann_vol > 0 else 0

        cumulative = (1 + returns).cumprod()
        rolling_max = cumulative.expanding().max()
        drawdown = cumulative / rolling_max - 1
        max_dd = drawdown.min()

        # Calmar比率
        calmar = ann_return / abs(max_dd) if max_dd != 0 else 0

        return {
            '年化收益率': f'{ann_return:.2%}',
            '年化波动率': f'{ann_vol:.2%}',
            '夏普比率': f'{sharpe:.2f}',
            '最大回撤': f'{max_dd:.2%}',
            'Calmar比率': f'{calmar:.2f}',
        }