缠论完整体系详解

# chan_minimal.py —— 缠论笔段中枢最小可运行版
# 依赖：pip install pandas numpy
import pandas as pd, numpy as np

# ---------- 1. K线包含处理 ----------
def merge_contain(k1, k2, direction):
    """k1前, k2后; direction: 'up'取高高 / 'down'取低低"""
    h1, l1 = k1; h2, l2 = k2
    if direction == 'up':
        return (max(h1, h2), max(l1, l2))
    else:
        return (min(h1, h2), min(l1, l2))

def remove_include(df: pd.DataFrame) -> pd.DataFrame:
    """df需有high/low列，返回去包含后的K线（带原索引）"""
    out, direction, last = [], None, None
    for i, row in df.iterrows():
        cur = (row['high'], row['low'])
        if last is None:
            last = cur; direction = None
            out.append((i, cur[0], cur[1]))
            continue
        hi, lo = cur
        # 判断包含
        if (hi <= last[0] and lo >= last[1]) or (hi >= last[0] and lo <= last[1]):
            # 有包含：定方向
            if direction is None:
                direction = 'up' if df.loc[i,'close'] >= df.loc[out[-1][0],'close'] else 'down'
            merged = merge_contain(last, cur, direction)
            last = merged
            out[-1] = (out[-1][0], merged[0], merged[1])
        else:
            # 无包含：更新方向
            direction = 'up' if hi > last[0] else 'down'
            last = cur
            out.append((i, cur[0], cur[1]))
    return pd.DataFrame(out, columns=['idx', 'high', 'low'])

# ---------- 2. 顶底分型 ----------
def find_fractals(kline: pd.DataFrame):
    """返回 list of (idx, price, type:'top'/'bottom')"""
    fracs = []
    h = kline['high'].values
    l = kline['low'].values
    idx = kline['idx'].values
    for i in range(1, len(kline)-1):
        if h[i] > h[i-1] and h[i] > h[i+1]:
            fracs.append((idx[i], h[i], 'top'))
        elif l[i] < l[i-1] and l[i] < l[i+1]:
            fracs.append((idx[i], l[i], 'bottom'))
    return fracs

# ---------- 3. 笔 ----------
def form_bi(fracs, min_k=3):
    """分型交替成笔，min_k: 两分型间最少隔几根K"""
    bis = []
    if len(fracs) < 2: return bis
    last = fracs[0]
    for f in fracs[1:]:
        if f[2] == last[2]: # 同类型分型，保留极值
            if (f[2]=='top' and f[1]>last[1]) or (f[2]=='bottom' and f[1]1]):
                last = f
            continue
        # 反向分型，成笔
        gap = abs(f[0] - last[0])
        if gap >= min_k:
            d = 1 if f[2]=='top' else -1
            bis.append((last[0], f[0], last[1], f[1], d))
            last = f
    return bis

# ---------- 4. 线段（特征序列法，67课） ----------
def form_segment(bis):
    """bis: form_bi输出。返回segments: (start_idx,end_idx,dir)"""
    segs = []
    if len(bis) < 3: return segs
    i = 0
    while i + 2 < len(bis):
        b1, b2, b3 = bis[i], bis[i+1], bis[i+2]
        # 三笔构成段：b1.up b2.down b3.up 且 b3.high > b1.high（向上段）
        if b1[4]==1 and b2[4]==-1 and b3[4]==1 and b3[3] > b1[3]:
            segs.append((b1[0], b3[1], 1))
            i += 2
        elif b1[4]==-1 and b2[4]==1 and b3[4]==-1 and b3[3] < b1[3]:
            segs.append((b1[0], b3[1], -1))
            i += 2
        else:
            i += 1
    return segs

# ---------- 5. 中枢（次级别三段重叠） ----------
def find_zhongshu(segs_with_prices):
    """segs_with_prices: [(start,end,dir,high,low),...]"""
    """返回中枢列表 [{zg,zd,start,end}]"""
    zs = []
    for i in range(len(segs_with_prices)-2):
        s1, s2, s3 = segs_with_prices[i], segs_with_prices[i+1], segs_with_prices[i+2]
        if s1[2] == s2[2] == s3[2]: # 同向段
            continue
        # zg = min(三段最高价), zd = max(三段最低价)
        zg = min(s1[3], s2[3], s3[3])
        zd = max(s1[4], s2[4], s3[4])
        if zg > zd: # 有重叠才构成中枢
            zs.append({'zg': zg, 'zd': zd,
                    'start': s1[0], 'end': s3[1]})
    return zs

# ---------- demo ----------
if __name__ == '__main__':
    np.random.seed(0)
    n = 200
    close = np.cumsum(np.random.randn(n)) + 100
    df = pd.DataFrame({'close': close})
    df['high'] = df['close'] + np.random.rand(n)*0.5
    df['low'] = df['close'] - np.random.rand(n)*0.5
    df = df.reset_index(drop=True)

    kl = remove_include(df)
    fr = find_fractals(kl)
    bi = form_bi(fr, min_k=3)
    seg = form_segment(bi)
    print(f"分型 {len(fr)} 个, 笔 {len(bi)} 根, 线段 {len(seg)} 段")

def detect_beichi(macd_areas, prices, threshold=0.8):
    """检测趋势背驰"""
    signals = []
    for i in range(1, len(macd_areas)):
        if prices[i] > prices[i-1] and macd_areas[i] < macd_areas[i-1] * threshold:
            signals.append({'type': 'top_beichi', 'index': i, 'ratio': macd_areas[i]/macd_areas[i-1]})
        elif prices[i] < prices[i-1] and macd_areas[i] < macd_areas[i-1] * threshold:
            signals.append({'type': 'bottom_beichi', 'index': i, 'ratio': macd_areas[i]/macd_areas[i-1]})
    return signals