From 5ad761fad44f9ef21216891a3f9a5994ff838be1 Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Mon, 25 May 2026 18:33:56 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=20ChanPivotClassifier:=20?= =?UTF-8?q?=E4=B8=AD=E6=9E=A2=E7=BB=93=E6=9E=84=E7=89=B9=E5=BE=81=E6=8F=90?= =?UTF-8?q?=E5=8F=96=20+=20=E6=A0=87=E7=AD=BE=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 训练数据集构建工具，从笔中枢提取 3 特征 (duration_norm, contraction, shift_norm) + 1 标签 (break_direction)。 Co-Authored-By: Claude Opus 4.6 --- ChanPivotClassifier.py | 258 ++++++++++++++++++++++++++++++++++++++ test_classifier.py | 272 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 530 insertions(+) create mode 100644 ChanPivotClassifier.py create mode 100644 test_classifier.py diff --git a/ChanPivotClassifier.py b/ChanPivotClassifier.py new file mode 100644 index 0000000..4b319d5 --- /dev/null +++ b/ChanPivotClassifier.py @@ -0,0 +1,258 @@ +""" +中枢结构特征提取 + 标签化 +Market Structure Dataset Builder — Phase 1 + +定位: 训练数据集构建工具，不是交易信号生成器。 +Feature 描述中枢内部结构，Label 记录中枢后实际演化。 +""" + +import math +import json +from ChanEnum import Chan_BI_DIR + + +class ChanPivotClassifier: + """ + 中枢结构特征提取 + 标签化 + 输入: bi_zs_list (list[ChanBIZS]) + 输出: 结构化数据集 (list[dict]) + """ + + DATASET_VERSION = "pivot_v1" + FEATURE_SCHEMA = ["duration_norm", "contraction", "shift_norm"] + LABEL_SCHEMA = {"name": "break_direction", "values": ["up", "down", "none"]} + + def __init__(self, bi_zs_list: list, symbol: str = "", timeframe: str = ""): + self.bi_zs_list = bi_zs_list + self.symbol = symbol + self.timeframe = timeframe + + # ------------------------------------------------------------------ + # Feature extraction + # ------------------------------------------------------------------ + + def _calc_duration(self, zs) -> int: + """持续时间: 第一笔首K → 最后一笔末K 的 index 差""" + bi_list = zs.bi_list + start_idx = bi_list[0].start_klc.index + end_idx = bi_list[-1].end_klc.index + return end_idx - start_idx + + def _calc_contraction(self, zs) -> float: + """收敛率: 后窗口振幅均值 / 前窗口振幅均值""" + bi_list = zs.bi_list + if len(bi_list) < 4: + return 1.0 + + n = min(3, len(bi_list) // 2) + first_ranges = [bi.high - bi.low for bi in bi_list[:n]] + last_ranges = [bi.high - bi.low for bi in bi_list[-n:]] + + first_mean = sum(first_ranges) / len(first_ranges) + last_mean = sum(last_ranges) / len(last_ranges) + + if first_mean == 0: + return 1.0 + return last_mean / first_mean + + def _calc_shift(self, zs) -> tuple[float, float]: + """重心漂移: 前后半段重心均值差 (原始值, 归一化值)""" + bi_list = zs.bi_list + mid = len(bi_list) // 2 + + first_centers = [(bi.high + bi.low) / 2 for bi in bi_list[:mid]] + last_centers = [(bi.high + bi.low) / 2 for bi in bi_list[mid:]] + + shift_raw = ( + sum(last_centers) / len(last_centers) + - sum(first_centers) / len(first_centers) + ) + + zs_height = zs.zg - zs.zd + if zs_height == 0: + shift_norm = 0.0 + else: + shift_norm = shift_raw / zs_height + + return shift_raw, shift_norm + + # ------------------------------------------------------------------ + # Label computation + # ------------------------------------------------------------------ + + @staticmethod + def _clamp(x: float, lo: float = 0.0, hi: float = 1.0) -> float: + return max(lo, min(hi, x)) + + def _compute_label(self, zs, contraction: float, shift_norm: float) -> dict: + """计算标签: up / down / none + 连续置信度""" + bi_out = zs.bi_out + + if bi_out is None: + return { + "label": "none", + "label_confidence": 0.0, + "label_detail": { + "bi_out_dir": "none", + "score_breakout": 0.0, + "score_shift": 0.0, + "score_contraction": 0.0, + }, + } + + zs_height = zs.zg - zs.zd + if zs_height == 0: + zs_height = 1e-8 + + # ---- 向上突破分数 ---- + if bi_out.dir == Chan_BI_DIR.UP: + raw_breakout = (bi_out.high - zs.gg) / zs_height + score_breakout_up = self._clamp(raw_breakout) + score_shift_up = math.tanh(self._clamp(shift_norm, -3.0, 3.0)) + score_contraction_up = max(0.0, 1.0 - contraction) + else: + score_breakout_up = 0.0 + score_shift_up = 0.0 + score_contraction_up = 0.0 + + up_score = ( + score_breakout_up * 0.5 + + score_shift_up * 0.3 + + score_contraction_up * 0.2 + ) + + # ---- 向下突破分数 ---- + if bi_out.dir == Chan_BI_DIR.DOWN: + raw_breakout = (zs.dd - bi_out.low) / zs_height + score_breakout_down = self._clamp(raw_breakout) + score_shift_down = math.tanh(self._clamp(-shift_norm, -3.0, 3.0)) + score_contraction_down = max(0.0, 1.0 - contraction) + else: + score_breakout_down = 0.0 + score_shift_down = 0.0 + score_contraction_down = 0.0 + + down_score = ( + score_breakout_down * 0.5 + + score_shift_down * 0.3 + + score_contraction_down * 0.2 + ) + + # ---- 判定 ---- + threshold = 0.15 + + if up_score > down_score and up_score > threshold: + label = "up" + confidence = up_score + detail = { + "bi_out_dir": "up", + "score_breakout": round(score_breakout_up, 4), + "score_shift": round(score_shift_up, 4), + "score_contraction": round(score_contraction_up, 4), + } + elif down_score > up_score and down_score > threshold: + label = "down" + confidence = down_score + detail = { + "bi_out_dir": "down", + "score_breakout": round(score_breakout_down, 4), + "score_shift": round(score_shift_down, 4), + "score_contraction": round(score_contraction_down, 4), + } + else: + label = "none" + confidence = max(up_score, down_score) + bi_dir = "up" if bi_out.dir == Chan_BI_DIR.UP else "down" + detail = { + "bi_out_dir": bi_dir, + "score_breakout": round(max(score_breakout_up, score_breakout_down), 4), + "score_shift": round(max(score_shift_up, score_shift_down), 4), + "score_contraction": round(max(score_contraction_up, score_contraction_down), 4), + } + + return { + "label": label, + "label_confidence": round(confidence, 4), + "label_detail": detail, + } + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def extract(self) -> list[dict]: + """主入口：对每个中枢提取 3 特征 + 1 标签""" + + # 第一遍：计算原始值 + raw = [] + for i, zs in enumerate(self.bi_zs_list): + if not zs.is_sure or len(zs.bi_list) < 3: + continue + + duration_raw = self._calc_duration(zs) + contraction = self._calc_contraction(zs) + shift_raw, shift_norm = self._calc_shift(zs) + + raw.append({ + "zs": zs, + "zs_index": i, + "duration_raw": duration_raw, + "contraction": contraction, + "shift_raw": shift_raw, + "shift_norm": shift_norm, + "zs_height": zs.zg - zs.zd, + }) + + # 归一化 duration: 除以均值 + if raw: + avg_duration = sum(r["duration_raw"] for r in raw) / len(raw) + else: + avg_duration = 1 + + # 第二遍：组装输出 + 计算 label + result = [] + for r in raw: + zs = r["zs"] + duration_norm = r["duration_raw"] / avg_duration if avg_duration > 0 else 1.0 + label_info = self._compute_label(zs, r["contraction"], r["shift_norm"]) + + # 时间处理 + start_time = None + end_time = None + if hasattr(zs, "start_time") and zs.start_time is not None: + start_time = str(zs.start_time) + if hasattr(zs, "end_time") and zs.end_time is not None: + end_time = str(zs.end_time) + + result.append({ + "dataset_version": self.DATASET_VERSION, + "feature_schema": self.FEATURE_SCHEMA, + "label_schema": self.LABEL_SCHEMA, + + "symbol": self.symbol, + "timeframe": self.timeframe, + "zs_index": r["zs_index"], + "zs_start_time": start_time, + "zs_end_time": end_time, + + "duration_norm": round(duration_norm, 4), + "contraction": round(r["contraction"], 4), + "shift_norm": round(r["shift_norm"], 4), + + "label": label_info["label"], + "label_confidence": label_info["label_confidence"], + "label_detail": label_info["label_detail"], + + "duration_raw": r["duration_raw"], + "shift_raw": round(r["shift_raw"], 6), + "zs_height": round(r["zs_height"], 6), + }) + + return result + + def export_json(self, path: str): + """导出为 JSON 文件""" + data = self.extract() + with open(path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, ensure_ascii=False, default=str) + return len(data) diff --git a/test_classifier.py b/test_classifier.py new file mode 100644 index 0000000..1b26570 --- /dev/null +++ b/test_classifier.py @@ -0,0 +1,272 @@ +""" +Phase 2: Run ChanPivotClassifier on real data, compute bi_out for each pivot, +export the dataset, and run single-variable statistics. + +Usage: python test_classifier.py +""" + +import csv +import sys +import os + +# Ensure Chan module is importable +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from TF_DF import TF_DF +from ChanPivotClassifier import ChanPivotClassifier +from ChanEnum import Chan_BI_DIR + +# Monkey-patch: TF_DF.init_TF_DF calls self.get_zs_list() which was removed. +# Add it back as an alias for get_bi_zs_list. +if not hasattr(TF_DF, 'get_zs_list'): + TF_DF.get_zs_list = lambda self, bi_list, seg_list: self.get_bi_zs_list(bi_list) + + +def load_csv(path: str) -> list[dict]: + """Load OHLCV CSV into list of dicts expected by TF_DF.""" + import pandas as pd + df = pd.read_csv(path) + df.columns = [c.lower() for c in df.columns] + # TF_DF expects 'date' column + if 'timestamp' in df.columns: + df.rename(columns={'timestamp': 'date'}, inplace=True) + df['date'] = pd.to_datetime(df['date']) + return df + + +def compute_bi_out(zs, bi_list: list) -> object: + """ + Determine the first bi after the pivot's end_bi that breaks out of the pivot range. + A breakout is: bi.high > zs.gg (up) or bi.low < zs.dd (down). + """ + if zs.end_bi is None or not zs.is_sure: + return None + + # Find end_bi position in bi_list + end_idx = None + for i, bi in enumerate(bi_list): + if bi is zs.end_bi or bi.index == zs.end_bi.index: + end_idx = i + break + + if end_idx is None: + return None + + # Look for the first bi after end_bi that breaks the pivot range + for i in range(end_idx + 1, len(bi_list)): + bi = bi_list[i] + if not bi.is_sure: + continue + # A breakout: goes above gg or below dd + if bi.high > zs.gg or bi.low < zs.dd: + return bi + + return None + + +def run_pipeline(csv_path: str, symbol: str, timeframe: str, interval: int = 1): + """Full pipeline: CSV → TF_DF → compute bi_out → ChanPivotClassifier.""" + print(f"\n{'='*60}") + print(f"Processing: {symbol} {timeframe}") + print(f"{'='*60}") + + # Step 1: Load data + df = load_csv(csv_path) + print(f"Loaded {len(df)} rows") + + # Step 2: Run TF_DF pipeline + tf_df = TF_DF(df, interval, timeframe) + print(f"KLC count: {len(tf_df.klc_list)}") + print(f"BI count: {len(tf_df.bi_list)}") + + # Get bi_zs_list via the seg-based method (matching find_all_bsp) + bi_zs_list = tf_df.cal_bi_zs(tf_df.seg_list) + print(f"Pivot count (raw): {len(bi_zs_list)}") + + # Filter to sure pivots with enough internal strokes + sure_pivots = [zs for zs in bi_zs_list if zs.is_sure and len(zs.bi_list) >= 3] + print(f"Pivot count (sure, >=3 strokes): {len(sure_pivots)}") + + # Step 3: Compute bi_out for each pivot + for zs in sure_pivots: + zs.bi_out = compute_bi_out(zs, tf_df.bi_list) + + bi_out_count = sum(1 for zs in sure_pivots if zs.bi_out is not None) + print(f"Pivots with bi_out: {bi_out_count}/{len(sure_pivots)}") + + # Step 4: Run ChanPivotClassifier + classifier = ChanPivotClassifier(sure_pivots, symbol=symbol, timeframe=timeframe) + dataset = classifier.extract() + print(f"Dataset samples: {len(dataset)}") + + # Step 5: Export + output_path = f"/tmp/chan_dataset_{symbol.replace('/', '_')}_{timeframe}.json" + count = classifier.export_json(output_path) + print(f"Exported {count} samples to {output_path}") + + return dataset + + +def run_statistics(dataset: list[dict]): + """Phase 2 statistics: single-variable analysis.""" + print(f"\n{'='*60}") + print("Phase 2 — Single-Variable Statistics") + print(f"{'='*60}\n") + + if not dataset: + print("No data to analyze.") + return + + total = len(dataset) + up = [d for d in dataset if d["label"] == "up"] + down = [d for d in dataset if d["label"] == "down"] + none_ = [d for d in dataset if d["label"] == "none"] + + print(f"Total samples: {total}") + print(f" Up: {len(up)} ({len(up)/total*100:.1f}%)") + print(f" Down: {len(down)} ({len(down)/total*100:.1f}%)") + print(f" None: {len(none_)} ({len(none_)/total*100:.1f}%)") + + # ================================================================ + # Feature 1: contraction vs break direction + # ================================================================ + print(f"\n--- Feature: contraction (convergence rate) ---") + for label, subset in [("up", up), ("down", down), ("none", none_)]: + if not subset: + continue + contractions = [d["contraction"] for d in subset] + avg = sum(contractions) / len(contractions) + print(f" {label}: mean contraction = {avg:.4f}") + + # Contraction < 0.7 → P(up)? + high_contraction = [d for d in dataset if d["contraction"] < 0.7] + if high_contraction: + up_in_hc = len([d for d in high_contraction if d["label"] == "up"]) + down_in_hc = len([d for d in high_contraction if d["label"] == "down"]) + print(f"\n Contraction < 0.7 (converging): {len(high_contraction)} samples") + print(f" P(up) = {up_in_hc/len(high_contraction)*100:.1f}%") + print(f" P(down) = {down_in_hc/len(high_contraction)*100:.1f}%") + + # Contraction > 1.2 → P(down)? + low_contraction = [d for d in dataset if d["contraction"] > 1.2] + if low_contraction: + up_in_lc = len([d for d in low_contraction if d["label"] == "up"]) + down_in_lc = len([d for d in low_contraction if d["label"] == "down"]) + print(f"\n Contraction > 1.2 (expanding): {len(low_contraction)} samples") + print(f" P(up) = {up_in_lc/len(low_contraction)*100:.1f}%") + print(f" P(down) = {down_in_lc/len(low_contraction)*100:.1f}%") + + # ================================================================ + # Feature 2: shift_norm vs break direction + # ================================================================ + print(f"\n--- Feature: shift_norm (center drift) ---") + for label, subset in [("up", up), ("down", down), ("none", none_)]: + if not subset: + continue + shifts = [d["shift_norm"] for d in subset] + avg = sum(shifts) / len(shifts) + print(f" {label}: mean shift_norm = {avg:.4f}") + + # shift > 0 → P(up)? + shift_up = [d for d in dataset if d["shift_norm"] > 0] + if shift_up: + up_in_su = len([d for d in shift_up if d["label"] == "up"]) + down_in_su = len([d for d in shift_up if d["label"] == "down"]) + print(f"\n shift_norm > 0 (drifting up): {len(shift_up)} samples") + print(f" P(up) = {up_in_su/len(shift_up)*100:.1f}%") + print(f" P(down) = {down_in_su/len(shift_up)*100:.1f}%") + + # shift < 0 → P(down)? + shift_down = [d for d in dataset if d["shift_norm"] < 0] + if shift_down: + up_in_sd = len([d for d in shift_down if d["label"] == "up"]) + down_in_sd = len([d for d in shift_down if d["label"] == "down"]) + print(f"\n shift_norm < 0 (drifting down): {len(shift_down)} samples") + print(f" P(up) = {up_in_sd/len(shift_down)*100:.1f}%") + print(f" P(down) = {down_in_sd/len(shift_down)*100:.1f}%") + + # ================================================================ + # Feature 3: duration_norm vs break direction + # ================================================================ + print(f"\n--- Feature: duration_norm (relative duration) ---") + for label, subset in [("up", up), ("down", down), ("none", none_)]: + if not subset: + continue + durations = [d["duration_norm"] for d in subset] + avg = sum(durations) / len(durations) + print(f" {label}: mean duration_norm = {avg:.4f}") + + # ================================================================ + # Combined: contraction < 0.7 AND shift_norm > 0 → P(up)? + # ================================================================ + print(f"\n--- Combined signals ---") + converging_up = [d for d in dataset if d["contraction"] < 0.7 and d["shift_norm"] > 0] + if converging_up: + up_in_cu = len([d for d in converging_up if d["label"] == "up"]) + down_in_cu = len([d for d in converging_up if d["label"] == "down"]) + print(f" Contraction < 0.7 AND shift_norm > 0: {len(converging_up)} samples") + print(f" P(up) = {up_in_cu/len(converging_up)*100:.1f}%") + print(f" P(down) = {down_in_cu/len(converging_up)*100:.1f}%") + + converging_down = [d for d in dataset if d["contraction"] < 0.7 and d["shift_norm"] < 0] + if converging_down: + up_in_cd = len([d for d in converging_down if d["label"] == "up"]) + down_in_cd = len([d for d in converging_down if d["label"] == "down"]) + print(f" Contraction < 0.7 AND shift_norm < 0: {len(converging_down)} samples") + print(f" P(up) = {up_in_cd/len(converging_down)*100:.1f}%") + print(f" P(down) = {down_in_cd/len(converging_down)*100:.1f}%") + + return dataset + + +def extract_symbol(csv_name: str) -> str: + """Extract symbol from filename like 'BTC_USDT_1d.csv'.""" + parts = csv_name.replace(".csv", "").split("_") + if len(parts) >= 2: + return f"{parts[0]}/{parts[1]}" + return csv_name + + +def extract_timeframe(csv_name: str) -> str: + """Extract timeframe from filename like 'BTC_USDT_1d.csv'.""" + parts = csv_name.replace(".csv", "").split("_") + if len(parts) >= 3: + return parts[2] + return "1d" + + +if __name__ == "__main__": + import glob + import json + + data_dir = "/Users/jack/Project/freqtrade/binance_data" + csv_files = sorted(glob.glob(f"{data_dir}/*_USDT_1h.csv")) + + if not csv_files: + print("No data files found.") + sys.exit(1) + + print(f"Found {len(csv_files)} data files:") + for f in csv_files: + print(f" {os.path.basename(f)}") + + # Batch process all coins + all_data = [] + for csv_path in csv_files: + basename = os.path.basename(csv_path) + symbol = extract_symbol(basename) + timeframe = extract_timeframe(basename) + try: + dataset = run_pipeline(csv_path, symbol, timeframe) + all_data.extend(dataset) + except Exception as e: + print(f" ERROR: {symbol} — {e}") + + # Export combined dataset + combined_path = "/tmp/chan_dataset_all_coins.json" + with open(combined_path, "w", encoding="utf-8") as f: + json.dump(all_data, f, indent=2, ensure_ascii=False, default=str) + print(f"\nCombined dataset: {len(all_data)} samples → {combined_path}") + + # Run statistics on combined dataset + run_statistics(all_data)