KDJ是什么？如何利用FX实时数据构建KDJ交易策略

import numpy as np
import pandas as pd

# ticks 需包含 timestamp、bid、ask 三列
ticks = ticks.copy()
ticks["timestamp"] = pd.to_datetime(
    ticks["timestamp"], utc=True, errors="coerce"
)
ticks["bid"] = pd.to_numeric(ticks["bid"], errors="coerce")
ticks["ask"] = pd.to_numeric(ticks["ask"], errors="coerce")

ticks = ticks.dropna(subset=["timestamp", "bid", "ask"])
ticks = ticks[
    (ticks["bid"] > 0)
    & (ticks["ask"] > 0)
    & (ticks["ask"] >= ticks["bid"])
]
ticks = ticks.drop_duplicates().sort_values("timestamp")

if ticks.empty:
    raise ValueError("没有可用的有效 Tick 数据")

ticks["mid"] = (ticks["bid"] + ticks["ask"]) / 2

# 生成以右侧时间标记的 5 分钟中间价 K 线
bars = (
    ticks.set_index("timestamp")["mid"]
    .resample("5min", label="right", closed="right")
    .ohlc()
)
bars.columns = ["open", "high", "low", "close"]
bars = bars.dropna()

if bars.empty:
    raise ValueError("无法生成有效的 5 分钟 K 线")

# 将时间间隔异常的区段分开，不自动填充缺失 K 线
bars["gap"] = bars.index.to_series().diff().gt(
    pd.Timedelta(minutes=5)
)
bars["segment"] = bars["gap"].cumsum()


def add_kdj(group):
    group = group.copy()

    low_n = group["low"].rolling(9, min_periods=9).min()
    high_n = group["high"].rolling(9, min_periods=9).max()
    price_range = (high_n - low_n).replace(0, np.nan)

    group["rsv"] = (
        (group["close"] - low_n) / price_range * 100
    )
    group["k"] = group["rsv"].ewm(
        alpha=1/3, adjust=False
    ).mean()
    group["d"] = group["k"].ewm(
        alpha=1/3, adjust=False
    ).mean()
    group["j"] = 3 * group["k"] - 2 * group["d"]
    group["ma20"] = group["close"].rolling(
        20, min_periods=20
    ).mean()

    cross_up = (
        (group["k"] > group["d"])
        & (group["k"].shift(1) <= group["d"].shift(1))
    )
    cross_down = (
        (group["k"] < group["d"])
        & (group["k"].shift(1) >= group["d"].shift(1))
    )

    # 示例规则：低位金叉 + 收盘价高于 MA20
    group["entry_signal"] = (
        cross_up
        & (group["k"] < 20)
        & (group["d"] < 20)
        & (group["close"] > group["ma20"])
    )

    # 示例规则：前一根 K 或 D 位于高位，随后出现死叉
    group["exit_signal"] = (
        cross_down
        & (
            (group["k"].shift(1) > 80)
            | (group["d"].shift(1) > 80)
        )
    )
    return group


bars = pd.concat(
    [add_kdj(group) for _, group in bars.groupby("segment", sort=False)]
).sort_index()

print(
    bars[["close", "k", "d", "j", "ma20",
          "entry_signal", "exit_signal"]].tail()
)