V 字翻转：形态识别与 AI 抢反弹

import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

def download_data(ticker, start='2020-02-01', end='2020-12-31'):
    return yf.download(ticker, start=start, end=end, progress=False)

def detect_v_reversal(df, lookback=20, drop_thresh=-0.15):
    """规则版 V 反识别：20日内跌15% + 站稳MA5/MA10 + 放量突破"""
    df = df.copy()
    df['MA5'] = df['Close'].rolling(5).mean()
    df['MA10'] = df['Close'].rolling(10).mean()
    df['Rolling_Max'] = df['Close'].rolling(lookback).max()
    df['Drawdown'] = df['Close'] / df['Rolling_Max'] - 1
    df['Vol_MA5'] = df['Volume'].rolling(5).mean()
    df['Vol_Ratio'] = df['Volume'] / df['Vol_MA5']

    signals = []
    for i in range(lookback + 10, len(df)):
        row = df.iloc[i]
        cond_drop = row['Drawdown'] <= drop_thresh          # 跌幅条件
        cond_ma   = row['Close'] > row['MA5'] > row['MA10'] # 站稳均线
        cond_vol  = row['Vol_Ratio'] > 1.5                  # 放量
        cond_yang = row['Close'] > row['Open']              # 阳线
        if cond_drop and cond_ma and cond_vol and cond_yang:
            signals.append({
                'date': df.index[i],
                'price': float(row['Close']),
                'drawdown': float(row['Drawdown']),
                'vol_ratio': float(row['Vol_Ratio']),
            })
    return pd.DataFrame(signals)

# 实战：检测 2020 年美股熔断 V 反
sp500 = download_data('^GSPC')
v_signals = detect_v_reversal(sp500)
print(f"检测到 {len(v_signals)} 个 V 反信号：")
print(v_signals)

# 可视化
sp500['Close'].plot(title='S&P 500 2020 V 反识别', figsize=(12, 5))
for _, sig in v_signals.iterrows():
    plt.axvline(sig['date'], color='red', linestyle='--', alpha=0.6)
    plt.annotate(f"买点
{sig['date'].date()}", xy=(sig['date'], sig['price']),
                 xytext=(10, 20), textcoords='offset points',
                 arrowprops=dict(arrowstyle='->', color='red'))
plt.show()

from xgboost import XGBClassifier
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import roc_auc_score
import yfinance as yf
import pandas as pd
import numpy as np

def build_features(df, vix_df=None):
    """构建 V 反预测特征"""
    df = df.copy()
    df['drawdown_5d']    = df['Close'] / df['Close'].shift(5) - 1
    df['drawdown_20d']   = df['Close'] / df['Close'].shift(20) - 1
    df['return']         = df['Close'].pct_change()
    df['volatility_10d'] = df['return'].rolling(10).std()
    df['vol_ma5']        = df['Volume'].rolling(5).mean()
    df['vol_ratio_3d']   = df['Volume'] / df['vol_ma5']
    df['vol_change']     = df['Volume'].pct_change(3)
    df['ma5']            = df['Close'].rolling(5).mean()
    df['close_ma5_ratio']= df['Close'] / df['ma5'] - 1
    delta = df['Close'].diff()
    gain  = delta.clip(lower=0).rolling(6).mean()
    loss  = (-delta.clip(upper=0)).rolling(6).mean()
    df['rsi_6'] = 100 - 100 / (1 + gain / (loss + 1e-9))
    if vix_df is not None:
        vix_aligned = vix_df['Close'].reindex(df.index, method='ffill')
        df['vix_level']  = vix_aligned
        df['vix_change'] = vix_aligned.pct_change(5)
    else:
        df['vix_level']  = 0
        df['vix_change'] = 0
    return df.dropna()

def create_label(df, forward_days=3, rebound_thresh=0.03):
    """标签：未来N日反弹超3%则标记为1（V反）"""
    future_max = df['Close'].rolling(forward_days).max().shift(-forward_days)
    future_return = future_max / df['Close'] - 1
    return (future_return >= rebound_thresh).astype(int)

def train_v_reversal_model(ticker='^GSPC', start='2015-01-01', end='2023-12-31'):
    df  = yf.download(ticker, start=start, end=end, progress=False)
    vix = yf.download('^VIX', start=start, end=end, progress=False)

    feat_df = build_features(df, vix)
    feat_df['label'] = create_label(feat_df)

    feature_cols = ['drawdown_5d', 'drawdown_20d', 'volatility_10d',
                    'vol_ratio_3d', 'vol_change', 'close_ma5_ratio',
                    'rsi_6', 'vix_level', 'vix_change']
    X = feat_df[feature_cols]
    y = feat_df['label']

    tscv = TimeSeriesSplit(n_splits=5)
    model = XGBClassifier(
        n_estimators=200, max_depth=4, learning_rate=0.05,
        scale_pos_weight=10,   # V 反是稀有事件，类别不平衡
        random_state=42, eval_metric='auc'
    )

    scores = []
    for train_idx, test_idx in tscv.split(X):
        model.fit(X.iloc[train_idx], y.iloc[train_idx])
        proba = model.predict_proba(X.iloc[test_idx])[:, 1]
        scores.append(roc_auc_score(y.iloc[test_idx], proba))

    print(f"时序 CV AUC: {scores.mean():.3f} ± {scores.std():.3f}")

    model.fit(X, y)
    latest_proba = model.predict_proba(X.iloc[-1:])[:, 1][0]
    print(f"当前 V 反概率: {latest_proba:.2%}")

    importance = pd.DataFrame({
        'feature': feature_cols,
        'importance': model.feature_importances_
    }).sort_values('importance', ascending=False)
    print("
特征重要性 Top 5:")
    print(importance.head())
    return model, latest_proba

if __name__ == '__main__':
    model, proba = train_v_reversal_model('^GSPC')