461 lines
15 KiB
Python
461 lines
15 KiB
Python
"""
|
|
前视偏差 (Look-Ahead Bias) 检测器
|
|
|
|
防止 AI agent 自动开发策略时引入前视偏差。两层防护:
|
|
|
|
1. 静态扫描 (AST): 扫描策略源码, 检测危险模式
|
|
- .shift(-N) 使用未来 bar
|
|
- df.iloc[i+N:] 切片未来数据
|
|
- close[-N] 负索引访问未来
|
|
- rolling(...).mean().shift(-1) 等
|
|
- 标准指标函数未来参数 (未来函数)
|
|
|
|
2. 动态验证 (运行时): 给策略喂"打乱后的未来", 看信号是否变化
|
|
- 修改 bar N+1..N+K 的 OHLC, 信号 entries[:N] 应保持不变
|
|
- 若变化 → 存在前视
|
|
|
|
用法:
|
|
from app.lookahead_check import check_strategy_file, check_strategy_code
|
|
|
|
# 静态扫描
|
|
report = check_strategy_file("strategies/my_strategy.py")
|
|
print(report.passed, report.issues)
|
|
|
|
# 动态验证
|
|
from app.lookahead_check import dynamic_check
|
|
result = dynamic_check(MyStrategy, df, params={"period": 14})
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
|
|
# ============================================================================
|
|
# 静态扫描规则
|
|
# ============================================================================
|
|
|
|
# 危险模式: (正则, 严重度, 描述)
|
|
# 严重度: HIGH=确定前视, MEDIUM=可疑, LOW=建议检查
|
|
STATIC_RULES = [
|
|
# ── HIGH: 确定使用未来数据 ──
|
|
(r"\.shift\s*\(\s*-\s*\d", "HIGH",
|
|
"使用 .shift(-N) 访问未来 bar, 这是明确的前视偏差"),
|
|
(r"\.iloc\s*\[\s*:.*[+].*\d\s*\]", "HIGH",
|
|
"iloc 切片包含未来索引"),
|
|
(r"\.loc\s*\[\s*:.*[+].*\d\s*\]", "HIGH",
|
|
"loc 切片包含未来索引"),
|
|
(r"\bclose\s*\[\s*-\s*\d", "HIGH",
|
|
"close 负索引访问 (numpy 从末尾取, 可能是未来)"),
|
|
(r"\bhigh\s*\[\s*-\s*\d", "HIGH",
|
|
"high 负索引访问"),
|
|
(r"\blow\s*\[\s*-\s*\d", "HIGH",
|
|
"low 负索引访问"),
|
|
(r"\bopen\s*\[\s*-\s*\d", "HIGH",
|
|
"open 负索引访问"),
|
|
(r"rolling\s*\([^)]*\)\s*\.\w+\s*\(\s*\)\s*\.shift\s*\(\s*-\s*\d", "HIGH",
|
|
"滚动统计后 shift 负值 (使用未来统计)"),
|
|
|
|
# ── MEDIUM: 可疑模式, 需人工确认 ──
|
|
(r"\.shift\s*\(\s*0\s*\)", "MEDIUM",
|
|
".shift(0) 无意义, 可能是 .shift(-N) 改写错误"),
|
|
(r"future|lookahead|tomorrow|next_bar|nextbar", "MEDIUM",
|
|
"代码中出现 future/lookahead 等关键词, 检查是否使用未来数据"),
|
|
(r"np\.roll\s*\([^,]+,\s*-?\d", "MEDIUM",
|
|
"np.roll 可能把末尾元素移到开头, 引入未来数据 (本项目已修复此 bug)"),
|
|
|
|
# ── LOW: 建议检查 ──
|
|
(r"\.values\[i\s*\+", "LOW",
|
|
"基于索引 i 访问 i+N, 确认 N 方向是过去而非未来"),
|
|
(r"df\[.+\]\.values\[i\s*\+", "LOW",
|
|
"基于索引 i 访问未来元素"),
|
|
]
|
|
|
|
|
|
# ============================================================================
|
|
# 检测结果
|
|
# ============================================================================
|
|
|
|
@dataclass
|
|
class Issue:
|
|
"""单个检测问题"""
|
|
line: int
|
|
col: int
|
|
severity: str # HIGH / MEDIUM / LOW
|
|
rule: str
|
|
message: str
|
|
code_snippet: str = ""
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"line": self.line,
|
|
"col": self.col,
|
|
"severity": self.severity,
|
|
"rule": self.rule,
|
|
"message": self.message,
|
|
"code_snippet": self.code_snippet,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class LookaheadReport:
|
|
"""前视偏差检测报告"""
|
|
file_path: str
|
|
issues: list = field(default_factory=list)
|
|
passed: bool = True
|
|
|
|
@property
|
|
def has_high(self) -> bool:
|
|
return any(i.severity == "HIGH" for i in self.issues)
|
|
|
|
@property
|
|
def has_medium(self) -> bool:
|
|
return any(i.severity == "MEDIUM" for i in self.issues)
|
|
|
|
def summary(self) -> str:
|
|
n_high = sum(1 for i in self.issues if i.severity == "HIGH")
|
|
n_med = sum(1 for i in self.issues if i.severity == "MEDIUM")
|
|
n_low = sum(1 for i in self.issues if i.severity == "LOW")
|
|
|
|
if not self.issues:
|
|
return (
|
|
f"前视偏差检测: ✓ 通过\n"
|
|
f" 未检测到前视风险模式"
|
|
)
|
|
|
|
status = "❌ 失败" if self.has_high else "⚠️ 警告"
|
|
lines = [
|
|
f"前视偏差检测: {status}",
|
|
f" HIGH (确定前视): {n_high}",
|
|
f" MEDIUM (可疑): {n_med}",
|
|
f" LOW (建议检查): {n_low}",
|
|
"",
|
|
"问题明细:",
|
|
]
|
|
for issue in self.issues:
|
|
icon = {"HIGH": "🔴", "MEDIUM": "🟡", "LOW": "🟢"}[issue.severity]
|
|
lines.append(
|
|
f" {icon} L{issue.line}: {issue.message}"
|
|
)
|
|
if issue.code_snippet:
|
|
lines.append(f" {issue.code_snippet}")
|
|
return "\n".join(lines)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"file_path": self.file_path,
|
|
"passed": self.passed,
|
|
"n_high": sum(1 for i in self.issues if i.severity == "HIGH"),
|
|
"n_medium": sum(1 for i in self.issues if i.severity == "MEDIUM"),
|
|
"n_low": sum(1 for i in self.issues if i.severity == "LOW"),
|
|
"issues": [i.to_dict() for i in self.issues],
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# 静态扫描
|
|
# ============================================================================
|
|
|
|
def check_strategy_code(code: str, file_path: str = "<string>") -> LookaheadReport:
|
|
"""
|
|
静态扫描策略源码, 检测前视偏差模式
|
|
|
|
参数:
|
|
code: Python 源码字符串
|
|
file_path: 文件路径 (用于报告)
|
|
|
|
返回:
|
|
LookaheadReport
|
|
"""
|
|
report = LookaheadReport(file_path=file_path)
|
|
lines = code.split("\n")
|
|
|
|
for line_no, line in enumerate(lines, 1):
|
|
# 跳过注释行
|
|
stripped = line.strip()
|
|
if stripped.startswith("#"):
|
|
continue
|
|
# 跳过 __future__ import (误报: 包含 "future" 关键词)
|
|
if stripped.startswith("from __future__") or stripped.startswith("import __future__"):
|
|
continue
|
|
|
|
for pattern, severity, message in STATIC_RULES:
|
|
for m in re.finditer(pattern, line, re.IGNORECASE):
|
|
col = m.start()
|
|
# 取上下文 (前后 20 字符)
|
|
ctx_start = max(0, col - 20)
|
|
ctx_end = min(len(line), m.end() + 20)
|
|
snippet = line[ctx_start:ctx_end].strip()
|
|
|
|
issue = Issue(
|
|
line=line_no,
|
|
col=col,
|
|
severity=severity,
|
|
rule=pattern,
|
|
message=message,
|
|
code_snippet=f"...{snippet}...",
|
|
)
|
|
report.issues.append(issue)
|
|
|
|
# 检测 AST 层面的危险: 赋值后用未来索引
|
|
try:
|
|
tree = ast.parse(code)
|
|
for node in ast.walk(tree):
|
|
# 检测 Subscript with negative index (如 close[-1])
|
|
if isinstance(node, ast.Subscript):
|
|
if isinstance(node.slice, ast.UnaryOp):
|
|
if isinstance(node.slice.op, ast.USub):
|
|
if isinstance(node.slice.operand, ast.Constant):
|
|
# 找到负索引, 但需确认是 close/high/low/open
|
|
if isinstance(node.value, ast.Name):
|
|
if node.value.id in ("close", "high", "low", "open"):
|
|
report.issues.append(Issue(
|
|
line=node.lineno,
|
|
col=node.col_offset,
|
|
severity="HIGH",
|
|
rule="ast:negative_index",
|
|
message=f"{node.value.id} 负索引访问 (可能是未来数据)",
|
|
))
|
|
except SyntaxError:
|
|
pass # 语法错误由其他工具报
|
|
|
|
# 判定通过/失败: 只要有 HIGH 就失败
|
|
report.passed = not report.has_high
|
|
return report
|
|
|
|
|
|
def check_strategy_file(file_path: str) -> LookaheadReport:
|
|
"""扫描策略文件"""
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
code = f.read()
|
|
return check_strategy_code(code, file_path)
|
|
|
|
|
|
def check_all_strategies(strategies_dir: str) -> list:
|
|
"""
|
|
扫描 strategies/ 目录下所有 .py 文件
|
|
|
|
返回:
|
|
[(file_path, report), ...]
|
|
"""
|
|
results = []
|
|
if not os.path.exists(strategies_dir):
|
|
return results
|
|
|
|
for fname in sorted(os.listdir(strategies_dir)):
|
|
if not fname.endswith(".py") or fname.startswith("_"):
|
|
continue
|
|
path = os.path.join(strategies_dir, fname)
|
|
report = check_strategy_file(path)
|
|
results.append((path, report))
|
|
|
|
return results
|
|
|
|
|
|
# ============================================================================
|
|
# 动态验证 — 修改未来 bar, 检查历史信号是否变化
|
|
# ============================================================================
|
|
|
|
def dynamic_check(
|
|
strategy_class,
|
|
df: pd.DataFrame,
|
|
params: Optional[dict] = None,
|
|
check_bars: int = 50,
|
|
perturb_range: int = 10,
|
|
) -> "DynamicCheckResult":
|
|
"""
|
|
动态前视检测: 修改 bar N+1..N+K 的 OHLC, 信号 entries[:N] 应不变
|
|
|
|
原理:
|
|
策略生成信号只用过去+当前数据, 所以修改未来 bar 不应影响历史信号。
|
|
若历史信号变化 → 存在前视。
|
|
|
|
参数:
|
|
strategy_class: 策略类
|
|
df: 原始数据
|
|
params: 策略参数 dict
|
|
check_bars: 检查前 N 根 bar 的信号是否变化
|
|
perturb_range: 修改未来多少根 bar
|
|
|
|
返回:
|
|
DynamicCheckResult
|
|
"""
|
|
params = params or {}
|
|
n = len(df)
|
|
if n < check_bars + perturb_range + 100:
|
|
return DynamicCheckResult(
|
|
passed=False,
|
|
reason=f"数据不足: {n} bars, 至少需要 {check_bars + perturb_range + 100}",
|
|
changed_bars=[],
|
|
)
|
|
|
|
# 1. 用原始数据生成基准信号
|
|
strategy = strategy_class(**params)
|
|
base_signals = strategy.generate_signals(df)
|
|
base_entries = base_signals.entries.copy()
|
|
base_exits = base_signals.exits.copy()
|
|
|
|
# 2. 修改未来 bar (check_bars 之后的 perturb_range 根)
|
|
df_perturbed = df.copy()
|
|
perturb_start = check_bars
|
|
perturb_end = min(check_bars + perturb_range, n)
|
|
|
|
# 显著修改未来 OHLC (±5%)
|
|
for col in ["open", "high", "low", "close"]:
|
|
if col in df_perturbed.columns:
|
|
original = df_perturbed[col].values.copy()
|
|
noise = np.random.uniform(0.95, 1.05, size=perturb_end - perturb_start)
|
|
original[perturb_start:perturb_end] *= noise
|
|
df_perturbed[col] = original
|
|
|
|
# 3. 用扰动后数据生成信号
|
|
strategy2 = strategy_class(**params)
|
|
perturbed_signals = strategy2.generate_signals(df_perturbed)
|
|
perturbed_entries = perturbed_signals.entries
|
|
perturbed_exits = perturbed_signals.exits
|
|
|
|
# 4. 比较前 check_bars 根的信号
|
|
changed_entries = np.where(
|
|
base_entries[:check_bars] != perturbed_entries[:check_bars]
|
|
)[0]
|
|
changed_exits = np.where(
|
|
base_exits[:check_bars] != perturbed_exits[:check_bars]
|
|
)[0]
|
|
|
|
changed_bars = sorted(set(changed_entries.tolist() + changed_exits.tolist()))
|
|
|
|
passed = len(changed_bars) == 0
|
|
if not passed:
|
|
reason = (
|
|
f"修改 bar {perturb_start}-{perturb_end} 后, "
|
|
f"前 {check_bars} 根 bar 中有 {len(changed_bars)} 根信号变化, "
|
|
f"存在前视偏差"
|
|
)
|
|
else:
|
|
reason = (
|
|
f"修改未来 {perturb_range} 根 bar 后, "
|
|
f"前 {check_bars} 根 bar 的信号无变化, 无前视偏差"
|
|
)
|
|
|
|
return DynamicCheckResult(
|
|
passed=passed,
|
|
reason=reason,
|
|
changed_bars=changed_bars,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class DynamicCheckResult:
|
|
"""动态前视检测结果"""
|
|
passed: bool
|
|
reason: str
|
|
changed_bars: list # 信号变化的 bar 索引
|
|
|
|
def summary(self) -> str:
|
|
icon = "✓" if self.passed else "✗"
|
|
lines = [
|
|
f"动态前视检测: {icon} {'通过' if self.passed else '失败'}",
|
|
f" {self.reason}",
|
|
]
|
|
if self.changed_bars:
|
|
lines.append(
|
|
f" 信号变化的 bar: {self.changed_bars[:10]}"
|
|
+ ("..." if len(self.changed_bars) > 10 else "")
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"passed": self.passed,
|
|
"reason": self.reason,
|
|
"changed_bars": self.changed_bars,
|
|
"n_changed": len(self.changed_bars),
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# 综合检查 (静态 + 动态)
|
|
# ============================================================================
|
|
|
|
def full_check(
|
|
strategy_class,
|
|
strategy_file: str,
|
|
df: pd.DataFrame,
|
|
params: Optional[dict] = None,
|
|
run_dynamic: bool = True,
|
|
) -> "FullCheckReport":
|
|
"""
|
|
综合前视检查: 静态扫描源码 + 动态验证信号
|
|
|
|
参数:
|
|
strategy_class: 策略类
|
|
strategy_file: 策略源码文件路径
|
|
df: 测试数据
|
|
params: 策略参数
|
|
run_dynamic: 是否运行动态检测 (默认 True)
|
|
|
|
返回:
|
|
FullCheckReport
|
|
"""
|
|
static_report = check_strategy_file(strategy_file)
|
|
|
|
dynamic_result = None
|
|
if run_dynamic:
|
|
try:
|
|
dynamic_result = dynamic_check(strategy_class, df, params)
|
|
except Exception as e:
|
|
dynamic_result = DynamicCheckResult(
|
|
passed=False,
|
|
reason=f"动态检测异常: {e}",
|
|
changed_bars=[],
|
|
)
|
|
|
|
return FullCheckReport(
|
|
static_report=static_report,
|
|
dynamic_result=dynamic_result,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class FullCheckReport:
|
|
"""综合检查报告"""
|
|
static_report: LookaheadReport
|
|
dynamic_result: Optional[DynamicCheckResult]
|
|
|
|
@property
|
|
def passed(self) -> bool:
|
|
if not self.static_report.passed:
|
|
return False
|
|
if self.dynamic_result and not self.dynamic_result.passed:
|
|
return False
|
|
return True
|
|
|
|
def summary(self) -> str:
|
|
lines = [
|
|
"=" * 60,
|
|
"前视偏差综合检测",
|
|
"=" * 60,
|
|
"",
|
|
self.static_report.summary(),
|
|
"",
|
|
]
|
|
if self.dynamic_result:
|
|
lines.append(self.dynamic_result.summary())
|
|
lines.append("")
|
|
lines.append(f"总结: {'✓ 通过' if self.passed else '❌ 未通过'}")
|
|
return "\n".join(lines)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"passed": self.passed,
|
|
"static": self.static_report.to_dict(),
|
|
"dynamic": self.dynamic_result.to_dict() if self.dynamic_result else None,
|
|
}
|