cyb50-quant/cat-fly/t1/kline_zone_analysis.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CYB50 K线 + 日线指标 舒适区 / 黑暗区 深度分析
──────────────────────────────────────────────────────────────────────
数据管道：
  cyb50_30min_2023_to_20260325.csv
    → resample 1D → 日线指标（RSI/KDJ/MACD/MA/BB/动量）

  backtest_vB_fee_*.csv（195笔，全部非死亡区，含30分钟K线指标）
    → attach_daily_indicators → 完整 30min + 日线 指标集

分析：
  单变量：每个指标分箱 → count / WR / avg_PnL / total_PnL
  双变量：关键指标对的 2D 组合矩阵
  区域识别：WR≥60% → 舒适区；WR≤35% → 黑暗区
"""
import sys, io, os, glob
if sys.platform == 'win32':
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')

import pandas as pd
import numpy as np
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

# ──────────────────────────────────────────────────────────────────
# 配置
# ──────────────────────────────────────────────────────────────────
DATA_CSV    = os.path.join(os.path.dirname(__file__), 'cyb50_30min_2023_to_20260325.csv')
MIN_TRADES  = 4      # 最少笔数（统计置信阈值）
WR_COMFORT  = 0.60   # 舒适区胜率阈值
WR_DARK     = 0.35   # 黑暗区胜率阈值
TOP_N       = 15     # 展示前 N 个区域
SEP         = '=' * 76


# ──────────────────────────────────────────────────────────────────
# 1. 数据加载
# ──────────────────────────────────────────────────────────────────
def load_backtest_vB() -> pd.DataFrame:
    """加载最新 backtest_vB_fee_*.csv（195笔非死亡区全量）"""
    pattern = os.path.join(os.path.dirname(__file__), 'backtest_vB_fee_*.csv')
    files = sorted(glob.glob(pattern))
    if not files:
        raise FileNotFoundError(
            "未找到 backtest_vB_fee_*.csv，请先运行 final_strategy.py")
    latest = files[-1]
    print(f"   回测数据: {os.path.basename(latest)}")
    df = pd.read_csv(latest, encoding='utf-8-sig')
    df['开仓时间'] = pd.to_datetime(df['开仓时间'])
    df['平仓时间'] = pd.to_datetime(df['平仓时间'])
    df['盈利']    = df['盈利标记'] == '盈'
    for c in ['实际盈亏', 'RSI', 'K', 'D', 'J', 'MACD_hist',
              'BB_width', 'ATR_Pct', 'Momentum', 'Volume_Ratio']:
        if c in df.columns:
            df[c] = pd.to_numeric(df[c], errors='coerce')
    return df


def build_daily_indicators() -> pd.DataFrame:
    """从30分钟K线聚合日线，计算日线技术指标"""
    raw = pd.read_csv(DATA_CSV, encoding='utf-8-sig')
    raw['DateTime'] = pd.to_datetime(raw['DateTime'])
    raw.set_index('DateTime', inplace=True)
    raw.sort_index(inplace=True)

    # 聚合为日线（只保留有数据的交易日）
    d = raw.resample('D').agg(
        Open=('Open', 'first'), High=('High', 'max'),
        Low=('Low', 'min'),    Close=('Close', 'last'),
        Volume=('Volume', 'sum')
    ).dropna(subset=['Close'])

    # ── RSI(14) ──
    delta   = d['Close'].diff()
    gain    = delta.where(delta > 0, 0).rolling(14).mean()
    loss    = (-delta.where(delta < 0, 0)).rolling(14).mean()
    d['RSI_D'] = 100 - 100 / (1 + gain / (loss + 1e-9))

    # ── KDJ(9,3,3) ──
    low9  = d['Low'].rolling(9).min()
    high9 = d['High'].rolling(9).max()
    rsv   = (d['Close'] - low9) / (high9 - low9 + 1e-9) * 100
    d['K_D'] = rsv.ewm(com=2, adjust=False).mean()
    d['D_D'] = d['K_D'].ewm(com=2, adjust=False).mean()
    d['J_D'] = 3 * d['K_D'] - 2 * d['D_D']

    # ── MACD(12,26,9) ──
    ema12 = d['Close'].ewm(span=12, adjust=False).mean()
    ema26 = d['Close'].ewm(span=26, adjust=False).mean()
    d['MACD_D']     = ema12 - ema26
    d['Signal_D']   = d['MACD_D'].ewm(span=9, adjust=False).mean()
    d['MACDhist_D'] = d['MACD_D'] - d['Signal_D']

    # ── 均线 & 偏离度 ──
    d['MA5_D']      = d['Close'].rolling(5).mean()
    d['MA10_D']     = d['Close'].rolling(10).mean()
    d['MA20_D']     = d['Close'].rolling(20).mean()
    d['MA60_D']     = d['Close'].rolling(60).mean()
    d['pct_MA20']   = (d['Close'] - d['MA20_D']) / d['MA20_D'] * 100
    d['pct_MA60']   = (d['Close'] - d['MA60_D']) / d['MA60_D'] * 100
    d['MA5_slope']  = d['MA5_D'].diff() / d['MA5_D'].shift() * 100  # MA5日斜率%

    # ── 布林带位置(20,2) ──
    bm     = d['Close'].rolling(20).mean()
    bstd   = d['Close'].rolling(20).std()
    bl, bu = bm - 2 * bstd, bm + 2 * bstd
    d['BB_pos_D'] = (d['Close'] - bl) / (bu - bl + 1e-9)   # 0~1

    # ── 动量 ──
    d['Mom5_D']  = d['Close'].pct_change(5)  * 100
    d['Mom10_D'] = d['Close'].pct_change(10) * 100

    return d


def attach_daily(trades: pd.DataFrame, daily: pd.DataFrame) -> pd.DataFrame:
    """为每笔交易附加开仓日的日线指标"""
    daily_cols = ['RSI_D', 'K_D', 'D_D', 'J_D',
                  'MACD_D', 'MACDhist_D',
                  'pct_MA20', 'pct_MA60', 'MA5_slope',
                  'BB_pos_D', 'Mom5_D', 'Mom10_D']
    result = trades.copy()
    for c in daily_cols:
        result[c] = float('nan')

    for i, row in trades.iterrows():
        entry_day = row['开仓时间'].normalize()
        if entry_day in daily.index:
            for c in daily_cols:
                if c in daily.columns:
                    result.at[i, c] = daily.loc[entry_day, c]
    return result


# ──────────────────────────────────────────────────────────────────
# 2. 指标分箱定义
# ──────────────────────────────────────────────────────────────────
IND_BINS = {
    # ── 30分钟指标 ──
    'RSI': (
        [-np.inf, 25, 35, 45, 55, 65, np.inf],
        ['超卖<25', '弱25-35', '偏弱35-45', '中45-55', '偏强55-65', '超买>65']
    ),
    'K': (
        [-np.inf, 20, 40, 60, 80, np.inf],
        ['超卖<20', '低20-40', '中40-60', '高60-80', '超买>80']
    ),
    'J': (
        [-np.inf, 0, 20, 50, 80, 100, np.inf],
        ['极卖<0', '超卖0-20', '低20-50', '高50-80', '超买80-100', '极买>100']
    ),
    'MACD_hist': (
        [-np.inf, -0.8, -0.2, 0, 0.2, 0.8, np.inf],
        ['强空<-0.8', '空-0.8~-0.2', '弱空~0', '弱多0~0.2', '多0.2~0.8', '强多>0.8']
    ),
    'Momentum': (
        [-np.inf, -0.03, -0.01, 0, 0.01, 0.03, np.inf],
        ['大跌>3%', '跌1-3%', '小跌<1%', '小涨<1%', '涨1-3%', '大涨>3%']
    ),
    'Volume_Ratio': (
        [-np.inf, 0.5, 0.8, 1.2, 1.8, np.inf],
        ['缩量<0.5', '偏低0.5-0.8', '平量0.8-1.2', '放量1.2-1.8', '大放>1.8']
    ),
    'BB_width': (
        [-np.inf, 0.025, 0.04, 0.06, 0.08, np.inf],
        ['极窄<2.5%', '窄2.5-4%', '中4-6%', '宽6-8%', '极宽>8%']
    ),
    'ATR_Pct': (
        [-np.inf, 0.006, 0.009, 0.013, 0.017, np.inf],
        ['极低<0.6%', '低0.6-0.9%', '中0.9-1.3%', '高1.3-1.7%', '极高>1.7%']
    ),
    # ── 日线指标 ──
    'RSI_D': (
        [-np.inf, 30, 40, 50, 60, 70, np.inf],
        ['超卖<30', '弱30-40', '偏弱40-50', '中50-60', '偏强60-70', '超买>70']
    ),
    'K_D': (
        [-np.inf, 20, 40, 60, 80, np.inf],
        ['超卖<20', '低20-40', '中40-60', '高60-80', '超买>80']
    ),
    'J_D': (
        [-np.inf, 0, 20, 50, 80, 100, np.inf],
        ['极卖<0', '超卖0-20', '低20-50', '高50-80', '超买80-100', '极买>100']
    ),
    'MACDhist_D': (
        [-np.inf, -2, -0.5, 0, 0.5, 2, np.inf],
        ['强空<-2', '空-2~-0.5', '弱空~0', '弱多0~0.5', '多0.5~2', '强多>2']
    ),
    'pct_MA20': (
        [-np.inf, -5, -2, 0, 2, 5, np.inf],
        ['大幅低<-5%', '低-5~-2%', '略低-2~0%', '略高0~2%', '高2~5%', '大幅高>5%']
    ),
    'pct_MA60': (
        [-np.inf, -8, -3, 0, 3, 8, np.inf],
        ['大幅低<-8%', '低-8~-3%', '略低-3~0%', '略高0~3%', '高3~8%', '大幅高>8%']
    ),
    'BB_pos_D': (
        [-np.inf, 0.1, 0.3, 0.5, 0.7, 0.9, np.inf],
        ['超下轨<0.1', '下轨0.1-0.3', '中下0.3-0.5', '中上0.5-0.7', '上轨0.7-0.9', '超上轨>0.9']
    ),
    'MA5_slope': (
        [-np.inf, -0.5, -0.1, 0.1, 0.5, np.inf],
        ['急降<-0.5%', '下降-0.5~-0.1%', '平-0.1~0.1%', '上升0.1~0.5%', '急升>0.5%']
    ),
    'Mom5_D': (
        [-np.inf, -4, -2, 0, 2, 4, np.inf],
        ['大跌>4%', '跌2-4%', '小跌<2%', '小涨<2%', '涨2-4%', '大涨>4%']
    ),
    'Mom10_D': (
        [-np.inf, -6, -3, 0, 3, 6, np.inf],
        ['大跌>6%', '跌3-6%', '小跌<3%', '小涨<3%', '涨3-6%', '大涨>6%']
    ),
}


def do_bin(series: pd.Series, col: str) -> pd.Series:
    """对指标列分箱，返回带标签的Categorical"""
    bins, labels = IND_BINS[col]
    return pd.cut(pd.to_numeric(series, errors='coerce'),
                  bins=bins, labels=labels, include_lowest=True)


# ──────────────────────────────────────────────────────────────────
# 3. 分析函数
# ──────────────────────────────────────────────────────────────────
def analyze_univariate(df: pd.DataFrame, min_n=MIN_TRADES) -> dict:
    """单变量分析：返回 {col: DataFrame(bin, n, wr, avg_pnl, total_pnl)}"""
    results = {}
    for col in IND_BINS:
        if col not in df.columns:
            continue
        binned = do_bin(df[col], col)
        tmp = df.copy()
        tmp['_b'] = binned
        grp = (tmp.groupby('_b', observed=True)
                  .agg(n=('盈利', 'count'),
                       wr=('盈利', 'mean'),
                       avg_pnl=('实际盈亏', 'mean'),
                       total_pnl=('实际盈亏', 'sum'))
                  .reset_index()
                  .rename(columns={'_b': 'bin'}))
        grp = grp[grp['n'] >= min_n].copy()
        grp['col'] = col
        results[col] = grp
    return results


# 双变量分析的关键指标对
BIVARIATE_PAIRS = [
    ('RSI',       'RSI_D'),        # 30min超卖 × 日线超卖共振
    ('J',         'J_D'),          # KDJ J值跨周期
    ('Momentum',  'Mom5_D'),       # 动量跨周期
    ('Volume_Ratio', 'ATR_Pct'),   # 放量 × 波动率
    ('MACD_hist', 'MACDhist_D'),   # MACD多空力道跨周期
    ('RSI',       'pct_MA20'),     # 超卖 × 均线偏离
    ('BB_pos_D',  'J_D'),          # 日线布林位置 × 日线KDJ
    ('RSI_D',     'Mom5_D'),       # 日线超卖 × 日线动量
    ('RSI',       'MA5_slope'),    # 30min超卖 × 日线趋势方向
    ('K',         'K_D'),          # KDJ K跨周期
]


def analyze_bivariate(df: pd.DataFrame,
                      pairs=BIVARIATE_PAIRS,
                      min_n=MIN_TRADES) -> dict:
    """双变量分析：返回 {c1×c2: DataFrame(c1_bin, c2_bin, n, wr, avg_pnl, total_pnl)}"""
    results = {}
    for c1, c2 in pairs:
        if c1 not in df.columns or c2 not in df.columns:
            continue
        tmp = df.copy()
        tmp['_b1'] = do_bin(df[c1], c1)
        tmp['_b2'] = do_bin(df[c2], c2)
        grp = (tmp.groupby(['_b1', '_b2'], observed=True)
                  .agg(n=('盈利', 'count'),
                       wr=('盈利', 'mean'),
                       avg_pnl=('实际盈亏', 'mean'),
                       total_pnl=('实际盈亏', 'sum'))
                  .reset_index())
        grp.columns = [c1, c2, 'n', 'wr', 'avg_pnl', 'total_pnl']
        grp = grp[grp['n'] >= min_n].copy()
        grp['pair'] = f'{c1}×{c2}'
        results[f'{c1}×{c2}'] = grp
    return results


# ──────────────────────────────────────────────────────────────────
# 4. 区域识别
# ──────────────────────────────────────────────────────────────────
def identify_zones(uni: dict, biv: dict):
    """
    从单变量+双变量结果中提炼舒适区（WR≥60%）和黑暗区（WR≤35%）。
    综合评分 = WR × avg_PnL（正向 → 舒适；负向 → 黑暗）
    """
    rows = []

    # 单变量
    for col, grp in uni.items():
        for _, r in grp.iterrows():
            rows.append({
                'type': '单变量',
                'dim1': col,
                'bin1': str(r['bin']),
                'dim2': '',
                'bin2': '',
                'n': int(r['n']),
                'wr': r['wr'],
                'avg_pnl': r['avg_pnl'],
                'total_pnl': r['total_pnl'],
                'score': r['wr'] * r['avg_pnl'],
            })

    # 双变量
    for key, grp in biv.items():
        c1, c2 = key.split('×')
        for _, r in grp.iterrows():
            rows.append({
                'type': '双变量',
                'dim1': c1,
                'bin1': str(r[c1]),
                'dim2': c2,
                'bin2': str(r[c2]),
                'n': int(r['n']),
                'wr': r['wr'],
                'avg_pnl': r['avg_pnl'],
                'total_pnl': r['total_pnl'],
                'score': r['wr'] * r['avg_pnl'],
            })

    all_df = pd.DataFrame(rows)
    comfort = (all_df[all_df['wr'] >= WR_COMFORT]
               .sort_values('score', ascending=False)
               .reset_index(drop=True))
    dark    = (all_df[all_df['wr'] <= WR_DARK]
               .sort_values('score', ascending=True)
               .reset_index(drop=True))
    return all_df, comfort, dark


# ──────────────────────────────────────────────────────────────────
# 5. 报告输出
# ──────────────────────────────────────────────────────────────────
def _wr_bar(wr: float, width=20) -> str:
    """ASCII 胜率条"""
    filled = int(wr * width)
    bar    = '█' * filled + '░' * (width - filled)
    return f'[{bar}] {wr:.1%}'


def print_univariate_summary(uni: dict):
    """打印每个指标的分箱胜率摘要（以WR排序）"""
    print(f'\n{"指标":<15} {"分箱":<20} {"笔数":>5} {"胜率":>28} {"均盈亏":>10} {"累计盈亏":>12}')
    print('-' * 92)
    for col, grp in uni.items():
        if grp.empty:
            continue
        grp_s = grp.sort_values('wr', ascending=False)
        first = True
        for _, r in grp_s.iterrows():
            col_disp = col if first else ''
            first = False
            bar = _wr_bar(r['wr'])
            pnl_sign = '+' if r['avg_pnl'] >= 0 else ''
            print(f'{col_disp:<15} {str(r["bin"]):<20} {int(r["n"]):>5}  '
                  f'{bar}  {pnl_sign}{r["avg_pnl"]:>9,.0f}  {r["total_pnl"]:>+12,.0f}')
        print()


def print_bivariate_heatmap(biv: dict, top_each=6):
    """打印每个双变量组合的 Top 舒适+黑暗区"""
    for pair_key, grp in biv.items():
        if grp.empty:
            continue
        c1, c2 = pair_key.split('×')
        print(f'\n  ◆ {pair_key}  （共{len(grp)}个有效组合，最少{MIN_TRADES}笔）')
        print(f'  {"":1} {c1:<22} {c2:<22} {"笔数":>4} {"胜率":>7} {"均盈亏":>10} {"累计盈亏":>12}')
        print(f'  {"-"} {"-"*22} {"-"*22} {"-"*4} {"-"*7} {"-"*10} {"-"*12}')
        # 舒适
        top_comfort = grp.nlargest(top_each, 'wr')
        for _, r in top_comfort.iterrows():
            flag = '★' if r['wr'] >= WR_COMFORT else ' '
            print(f'  {flag} {str(r[c1]):<22} {str(r[c2]):<22} {int(r["n"]):>4} '
                  f'{r["wr"]:>7.1%} {r["avg_pnl"]:>+10,.0f} {r["total_pnl"]:>+12,.0f}')
        # 黑暗（如果有）
        dark_rows = grp[grp['wr'] <= WR_DARK].nsmallest(min(3, top_each), 'wr')
        if not dark_rows.empty:
            print(f'  --- 黑暗 ---')
            for _, r in dark_rows.iterrows():
                print(f'  ✗ {str(r[c1]):<22} {str(r[c2]):<22} {int(r["n"]):>4} '
                      f'{r["wr"]:>7.1%} {r["avg_pnl"]:>+10,.0f} {r["total_pnl"]:>+12,.0f}')


def print_zones(comfort: pd.DataFrame, dark: pd.DataFrame, top_n=TOP_N):
    """打印最终舒适区/黑暗区汇总"""
    def _fmt_zone(row):
        if row['dim2']:
            return f"{row['dim1']}={row['bin1']}  &  {row['dim2']}={row['bin2']}"
        return f"{row['dim1']}={row['bin1']}"

    print(f'\n{"★ 舒适区 TOP":=<76}')
    print(f'  {"#":>3} {"类型":<5} {"条件":<52} {"笔数":>4} {"胜率":>7} {"均盈亏":>10}')
    print(f'  {"---":>3} {"-----":<5} {"-"*52} {"-"*4} {"-"*7} {"-"*10}')
    for i, (_, r) in enumerate(comfort.head(top_n).iterrows(), 1):
        cond = _fmt_zone(r)
        print(f'  {i:>3} {r["type"]:<5} {cond:<52} {int(r["n"]):>4} '
              f'{r["wr"]:>7.1%} {r["avg_pnl"]:>+10,.0f}')

    print(f'\n{"✗ 黑暗区 TOP":=<76}')
    print(f'  {"#":>3} {"类型":<5} {"条件":<52} {"笔数":>4} {"胜率":>7} {"均盈亏":>10}')
    print(f'  {"---":>3} {"-----":<5} {"-"*52} {"-"*4} {"-"*7} {"-"*10}')
    for i, (_, r) in enumerate(dark.head(top_n).iterrows(), 1):
        cond = _fmt_zone(r)
        print(f'  {i:>3} {r["type"]:<5} {cond:<52} {int(r["n"]):>4} '
              f'{r["wr"]:>7.1%} {r["avg_pnl"]:>+10,.0f}')


# ──────────────────────────────────────────────────────────────────
# 6. CSV 导出
# ──────────────────────────────────────────────────────────────────
def export_csv(trades: pd.DataFrame, all_zones: pd.DataFrame,
               comfort: pd.DataFrame, dark: pd.DataFrame):
    ts  = datetime.now().strftime('%Y%m%d_%H%M%S')
    out = os.path.dirname(__file__)

    # 每笔交易完整指标（含日线）
    fname_trades = os.path.join(out, f'zone_trades_{ts}.csv')
    trades.to_csv(fname_trades, index=False, encoding='utf-8-sig')
    print(f'   交易明细（含日线指标）: {os.path.basename(fname_trades)}  ({len(trades)}笔)')

    # 全量分析结果
    fname_zones = os.path.join(out, f'zone_analysis_{ts}.csv')
    all_zones.to_csv(fname_zones, index=False, encoding='utf-8-sig')
    print(f'   全量区域分析: {os.path.basename(fname_zones)}  ({len(all_zones)}行)')

    # 舒适区
    fname_c = os.path.join(out, f'zone_comfort_{ts}.csv')
    comfort.to_csv(fname_c, index=False, encoding='utf-8-sig')
    print(f'   舒适区: {os.path.basename(fname_c)}  ({len(comfort)}个区域)')

    # 黑暗区
    fname_d = os.path.join(out, f'zone_dark_{ts}.csv')
    dark.to_csv(fname_d, index=False, encoding='utf-8-sig')
    print(f'   黑暗区: {os.path.basename(fname_d)}  ({len(dark)}个区域)')


# ──────────────────────────────────────────────────────────────────
# 7. 主流程
# ──────────────────────────────────────────────────────────────────
def main():
    print(SEP)
    print('  CYB50 K线 + 日线指标  舒适区 / 黑暗区 深度分析')
    print(f'  Version B（非死亡区全量）| 最少笔数≥{MIN_TRADES} | '
          f'舒适≥{WR_COMFORT:.0%} | 黑暗≤{WR_DARK:.0%}')
    print(SEP)

    # ── 加载数据 ──
    print(f'\n📂 加载数据...')
    trades = load_backtest_vB()
    print(f'   回测交易: {len(trades)}笔  '
          f'{trades["开仓时间"].min().date()} ~ {trades["开仓时间"].max().date()}')

    print(f'\n📈 计算日线指标...')
    daily = build_daily_indicators()
    print(f'   日线: {len(daily)}条  {daily.index[0].date()} ~ {daily.index[-1].date()}')

    print(f'\n🔗 关联日线指标...')
    trades = attach_daily(trades, daily)
    daily_hits = trades['RSI_D'].notna().sum()
    print(f'   成功关联日线指标: {daily_hits}/{len(trades)} 笔')

    # ── 分析 ──
    print(f'\n🔬 单变量分析（{len(IND_BINS)}个指标）...')
    uni = analyze_univariate(trades)
    print(f'   完成: {len(uni)}个指标')

    print(f'\n🔬 双变量分析（{len(BIVARIATE_PAIRS)}个组合对）...')
    biv = analyze_bivariate(trades)
    print(f'   完成: {len(biv)}个组合对')

    print(f'\n🎯 识别舒适区 / 黑暗区...')
    all_zones, comfort, dark = identify_zones(uni, biv)
    print(f'   舒适区（WR≥{WR_COMFORT:.0%}）: {len(comfort)}个')
    print(f'   黑暗区（WR≤{WR_DARK:.0%}）: {len(dark)}个')

    # ── 报告 ──
    print(f'\n{SEP}')
    print('  单变量分析 — 各指标分箱胜率')
    print(SEP)
    print_univariate_summary(uni)

    print(f'\n{SEP}')
    print('  双变量分析 — 关键指标对组合')
    print(SEP)
    print_bivariate_heatmap(biv)

    print(f'\n{SEP}')
    print('  综合区域汇总')
    print(SEP)
    print_zones(comfort, dark)

    # ── 导出 ──
    print(f'\n📁 导出CSV...')
    export_csv(trades, all_zones, comfort, dark)

    print()
    print(SEP)
    print('  分析完成')
    print(SEP)


if __name__ == '__main__':
    main()