chore: 移除不再使用的 ChanMacro、system、tests。

这些目录已废弃,从仓库中清理。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
2026-08-05 18:11:29 +08:00
co-authored by Cursor
parent f2e77e1bdb
commit e2e45bc1bc
51 changed files with 0 additions and 6172 deletions
@@ -1,131 +0,0 @@
"""
validation/transition_validator.py — Validates regime stability.
Tests: Transition matrix, average duration, flip rate, state entropy.
Answers: "Does the regime design produce stable, persistent states?"
Hard requirements:
- avg_duration > 5 days
- flip_rate < 15%
- Fails → regime definition needs redesign.
"""
from typing import Optional
import sqlite3
import logging
import numpy as np
import pandas as pd
from config import config
from .metrics import transition_matrix, regime_duration_stats
logger = logging.getLogger(__name__)
class TransitionReport:
"""Structured report for regime stability validation."""
def __init__(self):
self.avg_duration: float = 0.0
self.flip_rate: float = 0.0
self.state_entropy: float = 0.0
self.n_days: int = 0
self.transition_matrix: Optional[pd.DataFrame] = None
self.persistence_score: float = 0.0
self.is_stable: bool = False
self.conclusion: str = ""
self.warnings: list[str] = []
def summary(self) -> str:
lines = [
f"Regime Stability (N={self.n_days} days)",
f" Avg Duration: {self.avg_duration:.1f} days (need > {config.regime_min_avg_duration})",
f" Flip Rate: {self.flip_rate:.1%} (need < {config.regime_max_flip_rate:.0%})",
f" State Entropy: {self.state_entropy:.3f}",
f" Persistence Score: {self.persistence_score:.2f}",
f" Stable: {'YES' if self.is_stable else 'NO — redesign needed'}",
]
if self.warnings:
lines.append(f" Warnings: {'; '.join(self.warnings)}")
if self.transition_matrix is not None:
lines.append(f" Transition Matrix:\n{self.transition_matrix.to_string()}")
lines.append(f"{self.conclusion}")
return "\n".join(lines)
class TransitionValidator:
"""
Validates regime temporal stability.
Regime must persist — not flip daily.
If flip_rate > 20% or avg_duration < 3 days → regime definition failed.
"""
def __init__(self, db_path: Optional[str] = None):
self.db_path = db_path or config.db_path
def validate(self, regime_labels: pd.Series) -> TransitionReport:
"""Validate a regime sequence for stability."""
report = TransitionReport()
report.n_days = len(regime_labels)
if len(regime_labels) < 30:
report.conclusion = "INSUFFICIENT DATA (< 30 days)"
return report
# Duration stats
stats = regime_duration_stats(regime_labels)
report.avg_duration = stats["avg_duration"]
report.flip_rate = stats["flip_rate"]
report.state_entropy = stats["state_entropy"]
# Transition matrix
report.transition_matrix = transition_matrix(regime_labels)
# Persistence: how often does regime stay the same?
diag = np.diag(report.transition_matrix.values)
report.persistence_score = round(float(np.mean(diag)), 2)
# Stability check
report.is_stable = (
report.avg_duration >= config.regime_min_avg_duration and
report.flip_rate <= config.regime_max_flip_rate
)
# Warnings
if report.avg_duration < 3:
report.warnings.append(f"CRITICAL: avg duration={report.avg_duration:.1f}d — regime flips too fast")
elif report.avg_duration < config.regime_min_avg_duration:
report.warnings.append(f"WARNING: avg duration={report.avg_duration:.1f}d < {config.regime_min_avg_duration}")
if report.flip_rate > 0.20:
report.warnings.append(f"CRITICAL: flip rate={report.flip_rate:.1%} — regime unstable")
elif report.flip_rate > config.regime_max_flip_rate:
report.warnings.append(f"WARNING: flip rate={report.flip_rate:.1%} > {config.regime_max_flip_rate:.0%}")
if report.state_entropy > 2.0:
report.warnings.append(f"NOTE: high state entropy={report.state_entropy:.2f}, regimes may be too fine-grained")
if report.is_stable:
report.conclusion = "PASS: regime design is stable"
else:
report.conclusion = "FAIL: regime definition needs adjustment"
return report
def validate_from_db(self) -> TransitionReport:
"""Load regime history from DB and validate stability."""
conn = sqlite3.connect(self.db_path)
df = pd.read_sql_query(
"SELECT date, regime FROM regime_history ORDER BY date", conn
)
conn.close()
if df.empty:
r = TransitionReport()
r.conclusion = "NO DATA"
return r
regimes = df.set_index("date")["regime"]
return self.validate(regimes)