jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加 ChanPivotClassifier: 中枢结构特征提取 + 标签化

Browse Source 
Phase 1 训练数据集构建工具，从笔中枢提取 3 特征 (duration_norm, contraction, shift_norm) + 1 标签 (break_direction)。

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jackyu66git
2026-05-25 18:33:56 +08:00
co-authored by Claude Opus 4.6
parent 9eae12f07d
commit 5ad761fad4
2 changed files with 530 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChanPivotClassifier.py
+258
View File
@@ -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)
test_classifier.py
+272
View File
@@ -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)