【Python技术】KDJ日线、周线、月线共振理论及简单示例

一、KDJ多周期共振理论

1. 共振现象的本质
   KDJ日线、周线、月线共振是指不同时间周期（日、周、月）的KDJ指标在同一时段内发出同步信号（如金叉/死叉、超买/超卖），表明市场在短、中、长期趋势上达成一致。这种现象通过叠加多周期数据，过滤单一周期的偶然性，增强趋势判断的可靠性。例如，月线代表长期方向，周线反映中期趋势，日线捕捉短期波动，三者的同步信号意味着市场力量高度集中。

   
   

2. 共振的判断标准

• 同步交叉信号
  ：日、周、月线KDJ同时出现金叉（K线上穿D线）或死叉（K线下穿D线）。
• 同向超买超卖
  ：三周期KDJ的J值均高于80（超买）或低于20（超卖），或D值分别突破80/20阈值。
• 趋势一致性
  ：三周期的K线和D线保持同方向运动（如同时向上或向下倾斜）。

二、买卖点判断策略

1. 买入信号

• 低位共振金叉
  ：当日、周、月线KDJ均出现金叉，且J值<20（超卖），价格接近支撑位时，视为强烈买入信号。例如，月线KDJ在低位金叉后，周线和日线同步金叉，往往预示趋势反转。
• 超卖区域共振
  ：若三周期KDJ的D值均低于20，且股价缩量企稳，可视为抄底机会。此时需结合成交量验证，若放量反弹则信号更可靠。
• 月线主导策略
  ：月线KDJ在低位金叉时，即使周线或日线尚未完全同步，也可逐步建仓，等待短期周期跟进。
  
  

• 高位共振死叉
  ：三周期KDJ同时死叉，且J值>80（超买），价格接近压力位时，提示趋势反转风险。若月线KDJ高位死叉，即使日线反弹也需警惕。
• 顶背离信号
  ：股价创新高但KDJ高点逐步下移（如月线J值走弱），尤其伴随周线死叉，可能引发大级别调整。
• 超买区域共振
  ：三周期D值均>80时，即使未死叉，也需减仓规避潜在回调。
• 
  

• 日线金叉但周/月线高位
  ：仅视为短线反弹，需快进快出。例如，日线金叉但周线J值>90，可能受制于中期压力。
• 周线反转中的日线波动
  ：若周线KDJ从高位回落但未死叉，日线金叉可能是主力洗盘后的二次拉升机会，需结合放量确认。

三、实战应用

1. 参数优化与公式整合

• 动态参数调整
  ：在震荡行情中缩短周期（如日线KDJ用6日参数），单边行情中延长周期（如周线用14日参数）。
• 多周期叠加公式
  ：在股票软件中编写综合指标（如加权平均三周期KDJ值），量化共振强度。例如，通达信公式可整合日、周、月KDJ值进行阈值报警。
  
  

• 结合MACD
  ：KDJ共振金叉时，若MACD同步金叉且柱状线翻红，可增强买入信号；反之，KDJ死叉与MACD死叉共振则强化卖出信号。
• 量价配合
  ：共振买点需成交量温和放大，卖点常伴随放量滞涨或缩量阴跌。例如，低位金叉+放量阳线=趋势启动确认。
• 均线系统辅助
  ：观察股价与MA60、MA120的关系。若KDJ共振金叉且股价站稳长期均线，趋势延续概率更高。
  
  

• 止损设置
  ：买入后以共振金叉当日最低价或前低下方3%-5%作为止损位；卖出信号触发后，反弹至压力位未突破则清仓。
• 分仓策略
  ：首次建仓30%-50%，确认趋势延续后加仓；强共振信号（如月线+周线+日线三重金叉）可提高至70%仓位，但需避免单一标的过度集中。

四、注意事项与局限性

1. 指标钝化与失效场景

• 单边行情钝化
  ：在长期上涨/下跌趋势中，KDJ可能持续超买/超卖而失效，此时需切换至趋势型指标（如MA或布林带）。
• 震荡市频繁交叉
  ：窄幅波动时KDJ易出现假信号，需结合波动率指标（如ATR）过滤无效交叉。

• 大周期主导原则
  ：月线>周线>日线。若月线KDJ高位死叉，即使日线金叉也需以风险控制为主。
• 适用性差异
  ：KDJ共振策略在主力控盘股、热门题材股中效果更佳；冷门股或流动性不足的标的易出现信号失真。

• 底背离
  ：股价创新低但KDJ低点抬高，可能预示反转，需等待日线放量金叉确认。
• 顶背离
  ：股价新高但KDJ高点下降，若周线同时出现顶背离，需果断减仓。

import streamlit as st
import akshare as ak
import pandas as pd
import numpy as np
import plotly.graph_objects as go
from datetime import datetime, timedelta


# 缓存数据获取（优化版）
def get_stock_data(symbol, start, end):
    try:
        code = f"{symbol}"
        # 获取数据
        df = ak.stock_zh_a_hist(
            symbol=code, period="daily",
            start_date=start.strftime("%Y%m%d"),
            end_date=end.strftime("%Y%m%d"


    
),
            adjust="qfq",
        )

        # 列名标准化
        df = df.rename(columns={
            '日期': 'date', '开盘': 'open', '收盘': 'close',
            '最高': 'high', '最低': 'low', '成交量': 'volume'
        })

        # 数据校验
        required_cols = ['date', 'open', 'close', 'high', 'low', 'volume']
        if not all(col in df.columns for col in required_cols):
            missing = set(required_cols) - set(df.columns)
            raise ValueError(f"缺失关键字段: {missing}")

        df['date'] = pd.to_datetime(df['date'])
        return df.set_index('date').sort_index()

    except Exception as e:
        st.error(f"数据获取失败：{str(e)}")
        st.info("常见原因：\n1. 股票代码需为6位数字\n2. 日期范围超出可获取范围\n3. 网络连接异常")
        return None


# 增强版KDJ计算（支持异常值过滤）
def calculate_kdj(df, n=9, m=3):
    df = df.copy()

    # 过滤异常值
    df = df[(df['high'] >= df['low']) &
            (df['close'].pct_change().abs() 0.2)]  # 过滤涨跌幅>20%的异常数据




    
    # KDJ计算逻辑
    low_min = df['low'].rolling(n, min_periods=1).min()
    high_max = df['high'].rolling(n, min_periods=1).max()
    rsv = (df['close'] - low_min) / (high_max - low_min + 1e-6) * 100  # 避免除零
    df['K'] = rsv.ewm(alpha=1 / m).mean()
    df['D'] = df['K'].ewm(alpha=1 / m).mean()
    df['J'] = 3 * df['K'] - 2 * df['D']

    return df[['K', 'D', 'J']]


# 多周期共振检测（增强信号过滤）
def detect_resonance(main_df, params):
    # 计算日线KDJ
    daily_kdj = calculate_kdj(main_df, n=params['daily_n'], m=params['m'])

    # 计算周线KDJ（带数据对齐）
    weekly_df = main_df.resample('W-FRI').agg({
        'open': 'first',
        'high': 'max',
        'low': 'min',
        'close': 'last'
    })
    weekly_kdj = calculate_kdj(weekly_df, n=params['weekly_n'], m=params['m'])
    weekly_kdj = weekly_kdj.resample('D').ffill().reindex(main_df.index, method='ffill')

    # 计算月线KDJ（带数据对齐）
    monthly_df = main_df.resample('M').agg({
        'open': 'first',
        'high': 'max',
        'low': 'min',
        'close': 'last'
    })
    monthly_kdj = calculate_kdj(monthly_df, n=params['monthly_n'], m=params['m'])
    monthly_kdj = monthly_kdj.resample('D').ffill().reindex(main_df.index, method='ffill')

    # 合并数据
    merged = pd.concat([
        daily_kdj.add_prefix('D_'),
        weekly_kdj.add_prefix('W_'),


    

        monthly_kdj.add_prefix('M_')
    ], axis=1).ffill()

    # 信号生成逻辑
    merged['D_金叉'] = (merged['D_K'] > merged['D_D']) & (merged['D_K'].shift() <= merged['D_D'].shift())
    merged['W_金叉'] = (merged['W_K'] > merged['W_D']) & (merged['W_K'].shift() <= merged['W_D'].shift())
    merged['M_金叉'] = (merged['M_K'] > merged['M_D']) & (merged['M_K'].shift() <= merged['M_D'].shift())

    # 三线共振信号（叠加过滤条件）
    merged['共振信号'] = (
            merged['D_金叉'] &
            merged['W_金叉'] &
            merged['M_金叉'] &
            (merged['D_K'] 50) &  # 日线处于低位
            (merged['W_J'] > merged['W_J'].shift(3)) &  # 周线J值趋势向上
            (merged['M_D'] 60)  # 月线D值未超买
    )

    return merged[['D_金叉', 'W_金叉', 'M_金叉', '共振信号']]


# 主界面布局（增强可视化）
def main():
    st.set_page_config(page_title="KDJ多周期共振系统", layout="wide")
    st.title("KDJ多周期共振选股系统")

    # 侧边栏参数设置
    with st.sidebar:
        st.header("⚙️参数设置")
        code = st.text_input("股票代码（如：600519）


    
", value="600519")
        start_date = st.date_input("开始日期", datetime.now() - timedelta(days=365))
        end_date = st.date_input("结束日期", datetime.now())

        st.subheader("KDJ参数")
        daily_n = st.slider("日线周期N", 3, 30, 5,
                            help="建议小于周线周期")
        weekly_n = st.slider("周线周期N", 10, 60, 27,
                             help="建议介于日线和月线之间")
        monthly_n = st.slider("月线周期N", 30, 120, 89,
                              help="建议大于周线周期")
        m = st.slider("平滑周期M", 2, 5, 3,
                      help="数值越小信号越敏感")

    # 获取数据
    df = get_stock_data(code, start_date, end_date)
    if df is None:
        return

    # 计算信号
    params = {'daily_n': daily_n, 'weekly_n': weekly_n, 'monthly_n': monthly_n, 'm': m}
    signals = detect_resonance(df, params)

    # 合并数据
    merged_df = df.join(signals).dropna()


    st.subheader(f"{code} KDJ共振分析图")

    # 创建增强版K线图
    fig = go.Figure()

    # K线主图
    fig.add_trace(go.Candlestick(
        x=merged_df.index,
        open=merged_df['open'],
        high=merged_df['high'],
        low=merged_df['low'],
        close=merged_df['close'],
        


    
name='K线',
        increasing_line_color='#FF4500',  # 自定义颜色
        decreasing_line_color='#4682B4'
    ))

    # 信号标记（增强可视化）
    signal_config = {
        'D_金叉': ('gold', 'circle', 1.02),
        'W_金叉': ('blue', 'square', 1.05),
        'M_金叉': ('purple', 'diamond', 1.08),
        '共振信号': ('lime', 'star', 1.12)
    }

    for signal, (color, symbol, offset) in signal_config.items():
        if merged_df[signal].any():
            dates = merged_df[merged_df[signal]].index
            fig.add_trace(go.Scatter(
                x=dates,
                y=merged_df.loc[dates, 'high'] * offset,
                mode='markers',
                marker=dict(
                    color=color,
                    size=16,
                    symbol=symbol,
                    line=dict(width=2, color='black')
                ),
                name=f'{signal}',
                hovertext=[f"价格：{merged_df.loc[d, 'close']:.2f}
日期：{d.strftime('%Y-%m-%d')}"
                           for d in dates]
            ))

    # 图表布局增强
    fig.update_layout(
        height=800,
        xaxis_rangeslider_visible=False,
        hovermode='x unified',
        legend=dict(orientation="h", yanchor="bottom"


    
, y=1.02),
        # plot_bgcolor='#1f2c38',  # 深色背景
        #paper_bgcolor='#1f2c38',
        font=dict(color='white'),
        margin=dict(l=20, r=20, t=40, b=20)
    )

    # 添加辅助线
    fig.update_yaxes(showgrid=True, gridwidth=0.5, gridcolor='#2d3a46')

    st.plotly_chart(fig, use_container_width=True





if __name__ == "__main__":
    main()