动量因子在中国A股市场的有效性检验：趋势跟踪与反转效应

import tushare as ts
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

# 设置tushare token
ts.set_token('YOUR_TUSHARE_TOKEN')
pro = ts.pro_api()

def get_stock_data(start_date='20100101', end_date='20251231'):
    """
    获取A股所有股票的历史数据
    """
    # 获取股票列表（剔除ST、退市）
    stock_list = pro.stock_basic(exchange='', list_status='L', fields='ts_code,name,industry')
    
    all_data = []
    for ts_code in stock_list['ts_code'].head(500):  # 先测500只，完整版用全部
        try:
            df = pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date)
            df['ts_code'] = ts_code
            all_data.append(df)
        except:
            continue
    
    # 合并数据
    data = pd.concat(all_data, ignore_index=True)
    data = data.sort_values(['ts_code', 'trade_date'])
    
    # 计算日收益率
    data['return'] = data.groupby('ts_code')['close'].pct_change()
    
    return data, stock_list

# 获取数据
data, stock_list = get_stock_data(start_date='20150101', end_date='20251231')
print(f"数据量：{len(data)} 条")

def calculate_momentum(data, lookback=252, holding=22):
    """
    计算动量因子
    
    Parameters:
    - lookback: 回溯期（交易日数，252≈1年）
    - holding: 持有期（交易日数，22≈1个月）
    """
    # 计算历史收益率（动量）
    data = data.copy()
    data['momentum'] = data.groupby('ts_code')['close'].pct_change(periods=lookback)
    
    # 去除NaN
    data = data.dropna(subset=['momentum'])
    
    return data

# 计算不同时间窗口的动量
for lookback in [22, 63, 126, 252]:  # 1个月, 3个月, 6个月, 1年
    data = calculate_momentum(data, lookback=lookback)
    print(f"Lookback={lookback}天，有效数据量：{data['momentum'].notna().sum()}")

def build_momentum_portfolio(data, date, n_groups=10):
    """
    构建动量组合（按动量打分分组）
    
    Parameters:
    - data: 数据集
    - date: 调仓日期
    - n_groups: 分组数（10→ Decile）
    
    Returns:
    - 各组平均收益率
    """
    # 筛选当前调仓日期的数据
    curr_data = data[data['trade_date'] == date].copy()
    
    if len(curr_data) < n_groups * 10:  # 数据太少，跳过
        return None
    
    # 按动量打分分组
    curr_data['group'] = pd.qcut(curr_data['momentum'], n_groups, labels=False)
    
    # 计算下期收益率（holding period）
    next_date = data[data['trade_date'] > date]['trade_date'].min()
    if pd.isna(next_date):
        return None
    
    next_data = data[data['trade_date'] == next_date]
    merged = curr_data[['ts_code', 'group']].merge(
        next_data[['ts_code', 'return']], on='ts_code', how='left'
    )
    
    # 各组平均收益率
    group_returns = merged.groupby('group')['return'].mean()
    
    return group_returns

# 回测：遍历所有调仓日期
dates = data['trade_date'].unique()
dates = sorted(dates)

results = []
for date in dates[252:]:  # 从第252天开始（需要足够历史数据）
    group_returns = build_momentum_portfolio(data, date, n_groups=10)
    if group_returns is not None:
        results.append(group_returns)

# 转换为DataFrame
results_df = pd.DataFrame(results)
results_df.index = dates[252:len(results_df)+252]

print(results_df.head())

def plot_momentum_results(results_df):
    """
    绘制动量因子表现
    """
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    
    # 1. 各组累计收益率
    cumulative_returns = (1 + results_df).cumprod() - 1
    axes[0, 0].plot(cumulative_returns.index, cumulative_returns[0], label='Group 1 (Low Momentum)', linewidth=2)
    axes[0, 0].plot(cumulative_returns.index, cumulative_returns[9], label='Group 10 (High Momentum)', linewidth=2)
    axes[0, 0].set_title('Cumulative Returns: Low vs High Momentum', fontsize=12)
    axes[0, 0].legend()
    axes[0, 0].grid(alpha=0.3)
    
    # 2. 多空组合收益率（High - Low）
    long_short = results_df[9] - results_df[0]
    cumulative_ls = (1 + long_short).cumprod() - 1
    axes[0, 1].plot(cumulative_ls.index, cumulative_ls, color='darkgreen', linewidth=2)
    axes[0, 1].set_title('Long-Short Momentum Portfolio', fontsize=12)
    axes[0, 1].grid(alpha=0.3)
    
    # 3. 各组平均收益率（横截面对比）
    avg_returns = results_df.mean() * 252  # 年化
    axes[1, 0].bar(range(10), avg_returns, color='skyblue', edgecolor='black')
    axes[1, 0].set_xlabel('Momentum Group (0=Low, 9=High)', fontsize=10)
    axes[1, 0].set_ylabel('Annualized Return', fontsize=10)
    axes[1, 0].set_title('Average Annualized Return by Group', fontsize=12)
    axes[1, 0].grid(alpha=0.3)
    
    # 4. 多空组合收益率分布
    axes[1, 1].hist(long_short, bins=50, density=True, alpha=0.7, color='lightcoral', edgecolor='black')
    axes[1, 1].axvline(long_short.mean(), color='red', linestyle='--', linewidth=2, label=f'Mean: {long_short.mean()*100:.2f}%')
    axes[1, 1].set_xlabel('Long-Short Return', fontsize=10)
    axes[1, 1].set_ylabel('Density', fontsize=10)
    axes[1, 1].set_title('Distribution of Long-Short Returns', fontsize=12)
    axes[1, 1].legend()
    axes[1, 1].grid(alpha=0.3)
    
    plt.tight_layout()
    plt.savefig('/Users/halo/workspace/astro-blog/public/images/2026-06-11-momentum-factor-china/momentum-backtest.png', dpi=300, bbox_inches='tight')
    plt.close()

plot_momentum_results(results_df)

def compare_momentum_windows(data):
    """
    对比不同回溯期（lookback）的动量表现
    """
    lookbacks = [22, 63, 126, 252]  # 1, 3, 6, 12个月
    results = {}
    
    for lookback in lookbacks:
        # 计算动量
        data = calculate_momentum(data, lookback=lookback)
        
        # 回测
        dates = data['trade_date'].unique()
        dates = sorted(dates)
        
        ls_returns = []
        for date in dates[lookback:]:
            group_returns = build_momentum_portfolio(data, date, n_groups=10)
            if group_returns is not None:
                ls_return = group_returns[9] - group_returns[0]
                ls_returns.append(ls_return)
        
        # 计算指标
        ls_returns = np.array(ls_returns)
        annual_return = ls_returns.mean() * 252
        annual_vol = ls_returns.std() * np.sqrt(252)
        sharpe = annual_return / annual_vol if annual_vol > 0 else 0
        
        results[f"{lookback}天"] = {
            '年化收益': f"{annual_return*100:.2f}%",
            '年化波动': f"{annual_vol*100:.2f}%",
            '夏普比率': f"{sharpe:.2f}"
        }
    
    return pd.DataFrame(results).T

comparison = compare_momentum_windows(data)
print(comparison)

def momentum_bull_vs_bear(data):
    """
    对比牛市和熊市的动量表现
    """
    # 定义牛熊市（用沪深300指数）
    index_data = pro.index_daily(ts_code='000300.SH', start_date='20150101', end_date='20251231')
    index_data = index_data.sort_values('trade_date')
    index_data['index_return'] = index_data['close'].pct_change()
    
    # 标记牛熊市（200日均线）
    index_data['ma200'] = index_data['close'].rolling(200).mean()
    index_data['market_state'] = np.where(index_data['close'] > index_data['ma200'], 'Bull', 'Bear')
    
    # 合并数据
    data = data.merge(index_data[['trade_date', 'market_state']], on='trade_date', how='left')
    
    # 分别计算牛市和熊市的动量收益
    bull_returns = []
    bear_returns = []
    
    for date in data['trade_date'].unique():
        curr_data = data[data['trade_date'] == date]
        if curr_data['market_state'].iloc[0] == 'Bull':
            group_returns = build_momentum_portfolio(curr_data, date)
            if group_returns is not None:
                bull_returns.append(group_returns[9] - group_returns[0])
        else:
            group_returns = build_momentum_portfolio(curr_data, date)
            if group_returns is not None:
                bear_returns.append(group_returns[9] - group_returns[0])
    
    bull_sharpe = np.mean(bull_returns) * 252 / (np.std(bull_returns) * np.sqrt(252))
    bear_sharpe = np.mean(bear_returns) * 252 / (np.std(bear_returns) * np.sqrt(252))
    
    return {
        '牛市夏普': f"{bull_sharpe:.2f}",
        '熊市夏普': f"{bear_sharpe:.2f}"
    }

market_comparison = momentum_bull_vs_bear(data)
print(market_comparison)

def hybrid_momentum_reversal(data, short_lookback=22, long_lookback=252):
    """
    混合策略：短期反转 + 长期动量
    """
    data = data.copy()
    
    # 计算短期收益率（反转）
    data['short_return'] = data.groupby('ts_code')['close'].pct_change(periods=short_lookback)
    
    # 计算长期收益率（动量）
    data['long_return'] = data.groupby('ts_code')['close'].pct_change(periods=long_lookback)
    
    # 打分：短期收益率越低越好（反转），长期收益率越高越好（动量）
    data['composite_score'] = -data['short_return'] + data['long_return']
    
    # 分组回测
    # ...（类似前面的代码）
    
    return data

hybrid_data = hybrid_momentum_reversal(data)

def industry_neutral_momentum(data):
    """
    行业中性化动量策略
    """
    data = data.copy()
    
    # 计算动量
    data = calculate_momentum(data, lookback=252)
    
    # 每个行业内部分组
    data['industry_group'] = data.groupby(['trade_date', 'industry'])['momentum'].transform(
        lambda x: pd.qcut(x, 10, labels=False)
    )
    
    # 跨行业组合（只买每个行业的高动量股票）
    long_stocks = data[data['industry_group'] == 9]
    
    return long_stocks

industry_neutral_data = industry_neutral_momentum(data)

class MomentumStockSelector:
    """
    A股动量选股策略
    """
    def __init__(self, lookback=252, holding=22, top_n=20):
        self.lookback = lookback
        self.holding = holding
        self.top_n = top_n  # 持仓数量
        
    def select_stocks(self, trade_date):
        """
        选股函数（每月调仓）
        """
        # 1. 获取数据
        data = self.get_stock_data(trade_date, lookback=self.lookback)
        
        # 2. 计算动量
        data['momentum'] = data.groupby('ts_code')['close'].pct_change(periods=self.lookback)
        
        # 3. 剔除ST、停牌、新股
        data = self.filter_stocks(data, trade_date)
        
        # 4. 行业中性化
        data = self.industry_neutralize(data)
        
        # 5. 选股（动量最高的top_n只）
        selected = data.nlargest(self.top_n, 'momentum')
        
        return selected['ts_code'].tolist()
    
    def filter_stocks(self, data, trade_date):
        """
        剔除不符合条件的股票
        """
        # 剔除ST
        data = data[~data['name'].str.contains('ST|退')]
        
        # 剔除停牌（当日成交量为0）
        data = data[data['vol'] > 0]
        
        # 剔除新股（上市不满252天）
        # ...（需获取上市日期）
        
        return data
    
    def industry_neutralize(self, data):
        """
        行业中性化
        """
        data['industry_rank'] = data.groupby('industry')['momentum'].rank(pct=True)
        data = data[data['industry_rank'] >= 0.9]  # 每个行业前10%
        
        return data
    
    def backtest(self, start_date='20150101', end_date='20251231'):
        """
        回测
        """
        dates = pd.date_range(start=start_date, end=end_date, freq='M')  # 每月调仓
        
        portfolio_returns = []
        for date in dates:
            selected_stocks = self.select_stocks(date.strftime('%Y%m%d'))
            
            # 等权持有
            next_date = date + pd.DateOffset(months=1)
            period_return = self.calculate_portfolio_return(selected_stocks, date, next_date)
            
            portfolio_returns.append(period_return)
        
        # 计算指标
        portfolio_returns = np.array(portfolio_returns)
        sharpe = portfolio_returns.mean() * 12 / (portfolio_returns.std() * np.sqrt(12))
        
        return {
            '年化收益': f"{portfolio_returns.mean() * 12 * 100:.2f}%",
            '夏普比率': f"{sharpe:.2f}",
            '最大回撤': f"{self.calculate_max_drawdown(portfolio_returns) * 100:.2f}%"
        }

# 使用
strategy = MomentumStockSelector(lookback=252, holding=22, top_n=20)
results = strategy.backtest(start_date='20180101', end_date='20251231')
print(results)

def risk_management(selected_stocks, max_position=0.05, stop_loss=-0.10):
    """
    风险管理模块
    
    Parameters:
    - max_position: 单只股票最大仓位（5%）
    - stop_loss: 止损线（-10%）
    """
    # 1. 仓位限制
    weights = {stock: min(1/len(selected_stocks), max_position) for stock in selected_stocks}
    
    # 2. 止损规则（需在实盘中监控）
    # if current_return < stop_loss:
    #     sell_stock(stock)
    
    # 3. 行业集中度限制
    # if industry_weight > 0.3:
    #     adjust_weights()
    
    return weights