"""
前视偏差 (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,
        }