KDJ 实战技巧：随机指标与 AI 增强结合

RSV = (Close - Low9) / (High9 - Low9) * 100
K_today = (K_yesterday * 2 + RSV_today) / 3
D_today = (D_yesterday * 2 + K_today) / 3
J_today = 3 * K_today - 2 * D_today

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

plt.rcParams['font.sans-serif'] = ['DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False

def calc_kdj(df, n=9, k_period=3, d_period=3):
    low_n = df['Low'].rolling(n, min_periods=1).min()
    high_n = df['High'].rolling(n, min_periods=1).max()
    rsv = (df['Close'] - low_n) / (high_n - low_n + 1e-9) * 100
    k = np.zeros(len(df))
    d = np.zeros(len(df))
    k[0] = 50.0
    d[0] = 50.0
    for i in range(1, len(df)):
        k[i] = (k[i-1] * (k_period-1) + rsv.iloc[i]) / k_period
        d[i] = (d[i-1] * (d_period-1) + k[i]) / d_period
    j = 3 * k - 2 * d
    return pd.DataFrame({'K': k, 'D': d, 'J': j, 'RSV': rsv.values}, index=df.index)


def traditional_kdj_strategy(df, kdj):
    """金叉 + J<10 双重过滤"""
    signals = pd.Series(0, index=df.index)
    k, d, j = kdj['K'], kdj['D'], kdj['J']
    # K 上穿 D 且 J 值处于低位（更可靠的买点）
    golden = (k > d) & (k.shift(1) <= d.shift(1)) & (j < 10)
    death = (k < d) & (k.shift(1) >= d.shift(1)) & (j > 90)
    signals[golden] = 1
    signals[death] = -1
    return signals


def backtest(df, signals, initial=1e6):
    position = 0
    cash = initial
    equity = []
    for i, (date, row) in enumerate(df.iterrows()):
        sig = signals.iloc[i]
        price = row['Close']
        if sig == 1 and position == 0:
            position = cash // price
            cash -= position * price
        elif sig == -1 and position > 0:
            cash += position * price
            position = 0
        equity.append(cash + position * price)
    return pd.Series(equity, index=df.index)


if __name__ == '__main__':
    # 贵州茅台 600519.SS（也可以换成 000001.SZ 平安银行、000858.SZ 五粮液）
    df = yf.download('600519.SS', start='2022-01-01', end='2024-12-31')
    df.columns = [c.lower() for c in df.columns]
    kdj = calc_kdj(df)
    sig = traditional_kdj_strategy(df, kdj)
    equity = backtest(df, sig)
    ret = equity.iloc[-1] / equity.iloc[0] - 1
    buy_hold = df['Close'].iloc[-1] / df['Close'].iloc[0] - 1
    print(f'传统 KDJ 收益: {ret*100:.2f}%   持有: {buy_hold*100:.2f}%')
    print(f'信号总数: {(sig!=0).sum()}')

from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import accuracy_score

def add_features(df):
    feat = pd.DataFrame(index=df.index)
    close = df['Close']
    # MACD 趋势确认
    ema12 = close.ewm(span=12).mean()
    ema26 = close.ewm(span=26).mean()
    dif = ema12 - ema26
    dea = dif.ewm(span=9).mean()
    feat['macd_dif'] = dif
    feat['macd_dea'] = dea
    feat['macd_hist'] = (dif - dea) * 2
    feat['macd_rising'] = (dif > dea).astype(int)
    # RSI 超买超卖二次确认
    delta = close.diff()
    gain = delta.clip(lower=0).rolling(14).mean()
    loss = (-delta.clip(upper=0)).rolling(14).mean()
    feat['rsi'] = 100 - 100 / (1 + gain / (loss + 1e-9))
    # 量价配合
    feat['vol_ratio'] = df['Volume'] / df['Volume'].rolling(20).mean()
    feat['vol_change'] = df['Volume'].pct_change(5)
    # 波动率与趋势强度
    feat['volatility'] = close.pct_change().rolling(10).std()
    feat['trend_5'] = close / close.rolling(5).mean() - 1
    feat['trend_20'] = close / close.rolling(20).mean() - 1
    # KDJ 自身特征
    feat['kdj_j'] = kdj['J']
    feat['kdj_kd_diff'] = kdj['K'] - kdj['D']
    feat['kdj_j_prev'] = kdj['J'].shift(1)
    return feat


def ai_enhanced_strategy(df, kdj, signals, feat):
    # 只在 KDJ 信号点采样
    sig_mask = signals != 0
    # 未来 5 日收益 > 1% 视为真信号（正例），<-1% 视为假信号（负例）
    fwd_ret = df['Close'].pct_change(5).shift(-5)
    y = pd.Series(0, index=df.index)
    y[fwd_ret > 0.01] = 1
    y[fwd_ret < -0.01] = -1
    valid = sig_mask & (y != 0)
    X_train = feat[valid].fillna(0)
    y_train = y[valid]
    # 时序交叉验证（防止未来数据泄露）
    model = RandomForestClassifier(n_estimators=120, max_depth=6,
                                   class_weight='balanced', random_state=42)
    tscv = TimeSeriesSplit(n_splits=5)
    scores = []
    for tr, te in tscv.split(X_train):
        model.fit(X_train.iloc[tr], y_train.iloc[tr])
        scores.append(accuracy_score(y_train.iloc[te], model.predict(X_train.iloc[te])))
    print(f'AI 模型时序 CV 准确率: {np.mean(scores)*100:.1f}%')
    # 全量训练后预测
    model.fit(X_train, y_train)
    pred = pd.Series(np.nan, index=df.index)
    pred.loc[sig_mask] = model.predict(feat[sig_mask].fillna(0))
    # 只保留模型确认的信号（KDJ 与 AI 同向）
    new_signals = signals.copy()
    keep = sig_mask & (pred == np.sign(signals))
    new_signals[~keep] = 0
    return new_signals

if __name__ == '__main__':
    feat = add_features(df)
    new_sig = ai_enhanced_strategy(df, kdj, sig, feat)
    ai_equity = backtest(df, new_sig)
    ai_ret = ai_equity.iloc[-1] / ai_equity.iloc[0] - 1
    print(f'AI 增强收益: {ai_ret*100:.2f}%')
    print(f'过滤后信号: {(new_sig!=0).sum()} / 原 {(sig!=0).sum()}')
    # 特征重要性
    print('Top 特征:', feat.columns[np.argsort(model.feature_importances_)[-5:]][::-1].tolist())