""" 中枢结构特征提取 + 标签化 Market Structure Dataset Builder — Phase 1 定位: 训练数据集构建工具，不是交易信号生成器。 Feature 描述中枢内部结构，Label 记录中枢后实际演化。 """ import math import json from typing import Optional from chanlun.core.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 # ------------------------------------------------------------------ @staticmethod def calc_duration(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 @staticmethod def calc_contraction(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 @staticmethod def calc_shift(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 @staticmethod def compute_duration_norm(duration_raw: int, historical_durations: list) -> float: """用历史窗口均值归一化 duration""" if not historical_durations: return 1.0 avg = sum(historical_durations) / len(historical_durations) if avg == 0: return 1.0 return duration_raw / avg @staticmethod def compute_features(zs, historical_durations: Optional[list] = None): """计算单个中枢的全部结构特征（实时友好）""" duration_raw = ChanPivotClassifier.calc_duration(zs) contraction = ChanPivotClassifier.calc_contraction(zs) shift_raw, shift_norm = ChanPivotClassifier.calc_shift(zs) if historical_durations is not None and len(historical_durations) > 0: duration_norm = ChanPivotClassifier.compute_duration_norm( duration_raw, historical_durations ) else: duration_norm = 1.0 return { "duration_raw": duration_raw, "duration_norm": round(duration_norm, 4), "contraction": round(contraction, 4), "shift_raw": round(shift_raw, 6), "shift_norm": round(shift_norm, 4), "zs_height": round(zs.zg - zs.zd, 6), } # ------------------------------------------------------------------ # 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 = ChanPivotClassifier.calc_duration(zs) contraction = ChanPivotClassifier.calc_contraction(zs) shift_raw, shift_norm = ChanPivotClassifier.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, }) # 第二遍：组装输出 + 计算 label result = [] for r in raw: zs = r["zs"] historical = [x["duration_raw"] for x in raw] duration_norm = ChanPivotClassifier.compute_duration_norm( r["duration_raw"], historical ) 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)