Python量化策略实验008：KDJ金叉死叉策略实战

大家好，我是量化老王。前面我们已经陆续搭建了4类核心量化策略：SMA纯趋势策略、双均线交叉策略、RSI超买超卖策略、MACD趋势背离策略，覆盖了“顺势跟随”“逆向布局”两种核心逻辑，适配趋势市、震荡市不同场景。今天实验008，我们开启一个全新的策略——KDJ金叉死叉策略，它结合了价格波动与强弱判断，通过K、D、J三条曲线的交叉信号捕捉趋势拐点，比单一均线策略更灵敏，比RSI策略更贴合趋势，是新手进阶的必备策略，也能与前文策略形成完美互补。

一、kdj策略核心逻辑与核心优势

KDJ指标（随机指标），核心是通过计算股价近期最高价、最低价与收盘价的相对位置，反映股价的强弱和超买超卖状态，核心由三条曲线组成：K线（快速线）、D线（慢速线）、J线（信号线），三者交叉形成的“金叉”“死叉”是策略的核心交易信号，逻辑比MACD简单、比SMA更具针对性。

与已写的4类策略相比，KDJ策略的核心优势的是“多信号共振”，既兼顾趋势方向，又能规避单一信号的虚假触发，具体区别一眼看懂（新手重点记）：

• 对比SMA/双均线策略：无需依赖价格与均线的位置关系，通过三条曲线交叉判断拐点，信号更灵敏，能更早捕捉趋势启动时机；
• 对比RSI策略：RSI侧重“超买超卖逆向布局”，KDJ侧重“趋势拐点顺势跟随”，更适配有明确短期趋势的行情，不易被震荡误导；
• 对比MACD策略：MACD滞后性较强，适合中长期趋势，KDJ灵敏度更高，适合中短期趋势捕捉，且计算逻辑更简单，新手易理解。

KDJ策略的核心逻辑（极简易懂，新手直接记）：

• 买入信号（金叉）：K线从下向上穿越D线，且穿越时J线同步向上突破50线（中性线），说明短期买盘崛起，趋势由弱转强，触发买入；
• 卖出信号（死叉）：K线从上向下跌穿D线，且穿越时J线同步向下跌破50线，说明短期卖盘主导，趋势走弱，触发卖出；
• 补充规则：J线突破80线为超买，跌破20线为超卖，超买超卖区间的金叉/死叉信号，可信度更高（无需额外优化，简单有效）。

延续前文一致的轻量化环境，无需新增复杂库，新手直接复用即可：Python 3.12+，核心库为pandas（数据处理+KDJ计算）、matplotlib（可视化）、akshare（获取A股行情数据），无需额外安装依赖，降低实操门槛。

2. 数据获取与预处理

本次依旧选用单只大盘蓝筹股（比亚迪，代码002594），获取2024年全年日线数据（前复权，消除除权除息对价格的影响），确保数据口径与前文策略一致，便于新手对比学习。KDJ指标默认选用行业通用参数（新手无需调整）：周期14日，K线平滑参数3，D线平滑参数3，无需额外拟合，直接复用即可。

# 导入核心库（与前文策略一致，无需新增）import akshare as akimport pandas as pdimport matplotlib.pyplot as plt# 1. 获取比亚迪2024年日线数据（前复权，含高低收等核心字段）df = ak.stock_zh_a_hist(symbol="002594", period="daily",start_date="20240101", end_date="20241231",adjust="qfq")# 2. 数据预处理（统一格式，避免计算失真，复用前文逻辑）df["日期"] = pd.to_datetime(df["日期"])  # 日期格式标准化df.set_index("日期", inplace=True)       # 以日期为索引df = df.sort_index()                    # 按日期升序排列（避免时间错乱）# 字段重命名为英文，简化后续代码编写df.rename(columns={"开盘":"open", "收盘":"close", "最高":"high", "最低":"low", "成交量":"volume"}, inplace=True)# 保留核心字段（无需多余字段，降低计算复杂度）df = df[["open", "close", "high", "low", "volume"]]print("数据预处理完成，预览前5行核心数据：")print(df[["close", "high", "low", "volume"]].head())

很多新手会直接调用第三方库的KDJ函数，容易陷入“黑箱操作”，今天我们手动计算KDJ的K、D、J三条曲线，掌握底层逻辑，同时生成金叉、死叉核心信号，严格遵循“不优化、不冗余”的新手友好原则。

先明确KDJ核心计算公式（极简版，新手无需死记，代码直接复用）：

• RSV = （当日收盘价 - 近14日最低价）÷（近14日最高价 - 近14日最低价）× 100（反映当日价格在近期波动中的位置）；
• K线 = RSV的3日移动平均（平滑RSV波动，避免信号频繁触发）；
• D线 = K线的3日移动平均（进一步平滑，降低虚假信号）；
• J线 = 3×K线 - 2×D线（信号线，反应信号灵敏度）。

# 1. 手动计算KDJ核心指标（14日周期，3日平滑，行业通用参数）# 计算近14日最高价、最低价df["high_14"] = df["high"].rolling(window=14).max()df["low_14"] = df["low"].rolling(window=14).min()# 计算RSV指标（避免分母为0，添加微小值）df["rsv"] = (df["close"] - df["low_14"]) / (df["high_14"] - df["low_14"] + 1e-8) * 100# 计算K线、D线（初始值取RSV的均值，避免前几行无数据）df["k"] = df["rsv"].rolling(window=3).mean()df["d"] = df["k"].rolling(window=3).mean()# 计算J线（核心信号线）df["j"] = 3 * df["k"] - 2 * df["d"]# 2. 生成KDJ金叉、死叉核心交易信号（无额外优化，新手直接复用）# 金叉（买入信号）：前一日K≤D，当日K>D，且J线突破50线df["golden_cross"] = (df["k"].shift(1) <= df["d"].shift(1)) & (df["k"] > df["d"]) & (df["j"] > 50)# 死叉（卖出信号）：前一日K≥D，当日Kdeath_cross"] = (df["k"].shift(1) >= df["d"].shift(1)) & (df["k"] < df["d"]) & (df["j"] < 50)# 统一信号格式：1=买入，0=卖出，None=无信号（便于后续回测）df["signal"] = Nonedf.loc[df["golden_cross"], "signal"] = 1df.loc[df["death_cross"], "signal"] = 0# 过滤无KDJ数据的行（前16行无完整KDJ数据，剔除）df_clean = df.dropna(subset=["k", "d", "j"])# 统计信号数量，直观查看信号触发情况print(f"\n2024年KDJ策略信号统计：")print(f"金叉（买入）信号数：{df_clean['golden_cross'].sum()}个")print(f"死叉（卖出）信号数：{df_clean['death_cross'].sum()}个")# 查看前10个有效交易信号详情print("\n交易信号示例（含KDJ数值）：")print(df_clean[df_clean["signal"].notna()][["close", "k", "d", "j", "signal"]].head(10))

延续前文新手友好原则，回测不考虑佣金、滑点，仅按当日收盘价成交，封装简洁回测函数，计算策略收益率、最大回撤两大核心指标，同时采集详细交易记录，让新手能清晰看到每一笔交易的盈亏情况，无需复杂计算。

# 封装回测函数（无佣金、无滑点，按收盘价成交，含交易记录）def backtest_kdj(df, signal_col, initial_capital=100000):df_backtest = df.copy()trade_records = []  # 存储交易记录share_num = 0       # 持股数量trade_id = 1        # 交易编号position = 0        # 持仓状态：0=空仓，1=满仓# 初始化资金相关字段df_backtest["cash"] = float(initial_capital)df_backtest["total_value"] = float(initial_capital)for i in range(len(df_backtest)):current_date = df_backtest.index[i]current_signal = df_backtest.iloc[i][signal_col]current_close = df_backtest.iloc[i]["close"]current_cash = df_backtest.iloc[i]["cash"]# 买入逻辑（空仓状态下触发买入信号）if current_signal == 1 and position == 0:position = 1buy_price = current_close# 计算可买股数（取整股，无佣金）share_num = current_cash // buy_pricetotal_cost = share_num * buy_priceremaining_cash = current_cash - total_cost# 更新资金字段df_backtest.iloc[i, df_backtest.columns.get_loc("cash")] = remaining_cash# 记录买入交易trade_records.append({"交易编号": trade_id,"交易类型": "买入","交易日期": current_date.strftime("%Y-%m-%d"),"成交价格": round(buy_price, 2),"成交股数": share_num,"交易金额": round(total_cost, 2),"剩余现金": round(remaining_cash, 2),"持仓状态": "满仓"trade_id += 1# 卖出逻辑（满仓状态下触发卖出信号）elif current_signal == 0 and position == 1:position = 0sell_price = current_closetotal_income = share_num * sell_price# 计算本次交易盈亏（卖出金额 - 买入成本）profit = total_income - (share_num * df_backtest.iloc[i-1]["close"])remaining_cash = current_cash + total_income# 更新资金字段df_backtest.iloc[i, df_backtest.columns.get_loc("cash")] = remaining_cash# 记录卖出交易trade_records.append({"交易编号": trade_id,"交易类型": "卖出","交易日期": current_date.strftime("%Y-%m-%d"),"成交价格": round(sell_price, 2),"成交股数": share_num,"交易金额": round(total_income, 2),"本次盈亏": round(profit, 2),"剩余现金": round(remaining_cash, 2),"持仓状态": "空仓"trade_id += 1share_num = 0  # 卖出后清空持股数量# 无信号时，延续前一日持仓和资金状态else:if i > 0:df_backtest.iloc[i, df_backtest.columns.get_loc("cash")] = df_backtest.iloc[i-1]["cash"]# 计算总资产（持仓时=现金+持股市值，空仓时=现金）if position == 1:total_value = df_backtest.iloc[i]["cash"] + share_num * current_closeelse:total_value = df_backtest.iloc[i]["cash"]df_backtest.iloc[i, df_backtest.columns.get_loc("total_value")] = total_value# 计算核心回测指标final_value = df_backtest.iloc[-1]["total_value"]total_return = (final_value - initial_capital) / initial_capital * 100# 计算最大回撤（风险指标，负值越小越稳健）df_backtest["max_value"] = df_backtest["total_value"].cummax()df_backtest["drawdown"] = (df_backtest["total_value"] - df_backtest["max_value"]) / df_backtest["max_value"] * 100max_drawdown = df_backtest["drawdown"].min()# 转换交易记录为DataFrame，便于展示trade_df = pd.DataFrame(trade_records) if trade_records else Nonereturn total_return, max_drawdown, df_backtest, trade_df# 执行KDJ策略回测kdj_return, kdj_drawdown, df_backtest, trade_df = backtest_kdj(df_clean, "signal")# 输出回测结果print("="*60)print("KDJ金叉死叉策略回测结果（2024年比亚迪，无佣金、无滑点）：")print(f"  总收益率：{kdj_return:.2f}%")print(f"  最大回撤：{kdj_drawdown:.2f}%")print(f"  最终总资产：{df_backtest.iloc[-1]['total_value']:.2f}元（初始100000元）")print("="*60)# 输出详细交易记录print("\n" + "="*80)print("KDJ策略详细交易记录：")print("="*80)if trade_df is not None and not trade_df.empty:print(trade_df.to_string(index=False))# 计算累计盈亏total_profit = trade_df[trade_df["交易类型"] == "卖出"]["本次盈亏"].sum()print(f"\n累计盈亏：{round(total_profit, 2)}元")else:print("无交易记录（未触发有效金叉/死叉信号）")print("="*80)

将股价、KDJ三条曲线、交易信号整合可视化，新手可以直观看到金叉、死叉信号的触发时机，以及信号与股价趋势的匹配度，快速判断策略的实战效果，同时熟悉KDJ指标的波动规律。

# 设置中文字体，避免乱码（复用前文逻辑，无需修改）plt.rcParams["font.sans-serif"] = ["SimHei"]plt.rcParams["axes.unicode_minus"] = False# 过滤无效数据，用于可视化df_viz = df_backtest.dropna(subset=["k", "d", "j"])# 创建上下分栏画布（上：股价+交易信号，下：KDJ三条曲线）fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8), gridspec_kw={"height_ratios": [3, 2]})# 上半部分：比亚迪收盘价 + 买入/卖出信号ax1.plot(df_viz["close"], color="blue", linewidth=1.5, label="比亚迪收盘价")# 买入信号（红色上箭头）ax1.scatter(df_viz[df_viz["signal"] == 1].index, df_viz[df_viz["signal"] == 1]["close"],color="red", marker="^", s=120, label="买入信号（金叉）", zorder=5)# 卖出信号（绿色下箭头）ax1.scatter(df_viz[df_viz["signal"] == 0].index, df_viz[df_viz["signal"] == 0]["close"],color="green", marker="v", s=120, label="卖出信号（死叉）", zorder=5)ax1.set_title("KDJ金叉死叉策略信号图（2024年比亚迪）", fontsize=14)ax1.set_ylabel("价格（元）", fontsize=12)ax1.legend(loc="upper left")ax1.grid(True, alpha=0.3)# 下半部分：KDJ三条曲线 + 超买超卖/中性线ax2.plot(df_viz["k"], color="red", linewidth=1.2, label="K线（快速线）")ax2.plot(df_viz["d"], color="blue", linewidth=1.2, label="D线（慢速线）")ax2.plot(df_viz["j"], color="orange", linewidth=1.2, label="J线（信号线）")ax2.axhline(y=80, color="gray", linestyle="--", alpha=0.5, label="超买线（80）")ax2.axhline(y=50, color="black", linestyle="--", alpha=0.5, label="中性线（50）")ax2.axhline(y=20, color="gray", linestyle="--", alpha=0.5, label="超卖线（20）")ax2.set_xlabel("日期", fontsize=12)ax2.set_ylabel("KDJ数值", fontsize=12)ax2.legend(loc="upper right")ax2.grid(True, alpha=0.3)# 调整布局，避免标签重叠plt.tight_layout()plt.show()# 输出策略关键统计信息signal_count = df_viz["signal"].notna().sum()buy_count = (df_viz["signal"] == 1).sum()sell_count = (df_viz["signal"] == 0).sum()print(f"\nKDJ策略关键统计：")print(f"总交易次数：{signal_count}次（买入{buy_count}次，卖出{sell_count}次）")print(f"平均持仓周期：{len(df_viz)/signal_count:.1f}天（粗略计算）")

本次搭建的KDJ金叉死叉策略，是完全区别于前文4类策略的全新逻辑，核心优势是“灵敏性+共振性”：通过K、D、J三条曲线交叉，结合J线与中性线的位置，双重验证信号有效性，既避免了SMA策略的滞后性，又减少了RSI策略的虚假信号，适配中短期趋势行情，新手可直接复用代码、复用参数，无需复杂调整。

该策略的核心价值的是“新手进阶适配”：计算逻辑比MACD简单，信号比SMA更明确，同时能与前文策略形成互补——趋势市可用SMA/MACD策略，震荡市可用RSI策略，中短期趋势市可用KDJ策略，覆盖更多市场场景，逐步搭建新手的量化策略体系。

2. 后续优化方向（不增加复杂度，新手可落地）

• 参数微调：测试9日、20日KDJ周期，对比不同周期的信号灵敏度，适配不同波动强度的个股；
• 信号过滤：仅保留“超买超卖区间”的信号（J线>80金叉、J线<20死叉），进一步减少虚假信号；
• 策略组合：与SMA策略共振（KDJ金叉+股价突破SMA），提升买入信号可信度，降低风险。

你跑代码后，KDJ策略的收益率和最大回撤是多少？对比前文的SMA、RSI、MACD策略，你觉得KDJ策略的实操难度更低吗？评论区晒出你的回测结果和交易记录！关注我，100篇Python量化实验干货持续更新，从单一策略到策略组合，帮你从量化新手成长为实操高手！