MyStrategy

"""Freqtrade 策略代码静态校验。

LLM 生成的代码可能少方法、少 import、用未来函数等。这里做静态检查：

- Python 语法合法（ast.parse 不抛）
- 含 IStrategy 子类
- 三个必需方法（populate_indicators / populate_entry_trend / populate_exit_trend）签名正确
- 关键属性：timeframe / stoploss / minimal_roi
- 未来函数嫌疑：``dataframe[...].iloc[-1]`` 这类直接读最后一行的写法
- 不调用聚宽 API（这是 Freqtrade 策略，不该出现 get_price 等）
"""
from __future__ import annotations

import ast
import re
from dataclasses import dataclass, field
from typing import List


@dataclass
class ValidationIssue:
    severity: str        # "error" | "warning"
    code: str
    message: str
    line: int = 0


@dataclass
class ValidationReport:
    code: str
    issues: List[ValidationIssue] = field(default_factory=list)

    @property
    def errors(self) -> List[ValidationIssue]:
        return [i for i in self.issues if i.severity == "error"]

    @property
    def warnings(self) -> List[ValidationIssue]:
        return [i for i in self.issues if i.severity == "warning"]

    @property
    def passed(self) -> bool:
        return len(self.errors) == 0

    def add(self, severity: str, code: str, message: str, line: int = 0) -> None:
        self.issues.append(ValidationIssue(severity=severity, code=code,
                                           message=message, line=line))


_REQUIRED_METHODS = {
    "populate_indicators",
    "populate_entry_trend",
    "populate_exit_trend",
}

# 已废弃的 v2 接口方法 → 现在 v3 用的新方法
_DEPRECATED_METHODS = {
    "populate_buy_trend": "populate_entry_trend",
    "populate_sell_trend": "populate_exit_trend",
}

_REQUIRED_CLASS_ATTRS = {"timeframe", "stoploss", "minimal_roi"}

_HALLUCINATED_APIS = {
    "get_price",
    "attribute_history",
    "history",
    "get_current_data",
    "order_target",
    "order_value",
    "context.portfolio",
}


def validate(code: str) -> ValidationReport:
    """对一段 Freqtrade 策略代码做静态校验。"""
    report = ValidationReport(code=code)

    # 1. 语法合法
    try:
        tree = ast.parse(code)
    except SyntaxError as e:
        report.add("error", "FT000",
                   f"Python 语法错误：{e.msg}", line=e.lineno or 0)
        return report

    # 2. 找 IStrategy 子类
    strategy_class = None
    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef):
            for base in node.bases:
                if (isinstance(base, ast.Name) and base.id == "IStrategy") or \
                   (isinstance(base, ast.Attribute) and base.attr == "IStrategy"):
                    strategy_class = node
                    break
            if strategy_class:
                break

    if strategy_class is None:
        report.add("error", "FT001",
                   "找不到 IStrategy 子类。Freqtrade 策略必须 class MyStrategy(IStrategy):")
        return report

    # 3. 检查必需方法（同步/异步都算）
    methods_found = {
        item.name for item in strategy_class.body
        if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef))
    }

    # 3a. 先看是不是用了已废弃的 v2 接口（populate_buy/sell_trend）。
    #     如果用了，给针对性的 FT010，并把对应的新方法从"缺失"里豁免，
    #     免得又泛泛报一遍 FT002 让人摸不着头脑。
    suppress_missing = set()
    for old_name, new_name in _DEPRECATED_METHODS.items():
        if old_name in methods_found:
            report.add("error", "FT010",
                       f"用了已废弃的 v2 接口方法 {old_name}；"
                       f"Freqtrade v3（INTERFACE_VERSION=3）应改名为 {new_name}",
                       line=strategy_class.lineno)
            suppress_missing.add(new_name)

    missing = _REQUIRED_METHODS - methods_found - suppress_missing
    for m in missing:
        report.add("error", "FT002",
                   f"缺必需方法 {m}（IStrategy 三个核心方法之一）",
                   line=strategy_class.lineno)

    # 4. 检查必需类属性
    class_attrs = set()
    for item in strategy_class.body:
        if isinstance(item, ast.Assign):
            for target in item.targets:
                if isinstance(target, ast.Name):
                    class_attrs.add(target.id)
        elif isinstance(item, ast.AnnAssign):
            if isinstance(item.target, ast.Name):
                class_attrs.add(item.target.id)

    missing_attrs = _REQUIRED_CLASS_ATTRS - class_attrs
    for a in missing_attrs:
        sev = "error" if a == "stoploss" else "warning"
        report.add(sev, "FT003",
                   f"缺类属性 {a}（Freqtrade 需要）",
                   line=strategy_class.lineno)

    # 5. 检查未来函数嫌疑（dataframe['col'].iloc[-1] 这种读最后一行的写法）
    iloc_pattern = re.compile(r"dataframe\[['\"][^'\"]+['\"]\]\.iloc\[-1\]")
    for i, line in enumerate(code.splitlines(), start=1):
        if iloc_pattern.search(line):
            report.add("warning", "FT004",
                       "嫌疑未来函数：直接读 dataframe.iloc[-1] 在 Freqtrade 是合法但容易"
                       "误用；应该用整列条件赋值 dataframe.loc[condition, 'enter_long']=1",
                       line=i)

    # 6. 检查不该出现的聚宽 API
    for api in _HALLUCINATED_APIS:
        for i, line in enumerate(code.splitlines(), start=1):
            if api in line and not line.strip().startswith("#"):
                report.add("error", "FT005",
                           f"不该出现 {api}（聚宽 API，不是 Freqtrade）",
                           line=i)
                break

    # 7. 检查 stoploss 是负数（Assign 和带注解的 AnnAssign 都查）
    def _check_stoploss(value: ast.expr, lineno: int) -> None:
        if isinstance(value, ast.UnaryOp) and isinstance(value.op, ast.USub):
            return  # -0.05，OK
        if (isinstance(value, ast.Constant)
                and isinstance(value.value, (int, float))
                and not isinstance(value.value, bool)
                and value.value >= 0):
            report.add("error", "FT006",
                       f"stoploss 必须是负数（如 -0.05 表示 -5%），实际 {value.value}",
                       line=lineno)

    # 8. 检查 minimal_roi：必须是含 "0" 键的字面量 dict，键应为字符串
    def _check_minimal_roi(value: ast.expr, lineno: int) -> None:
        if not isinstance(value, ast.Dict):
            return  # 用变量/函数构造的，静态判断不了，跳过
        key_consts = [
            k.value for k in value.keys
            if isinstance(k, ast.Constant)
        ]
        has_int_key = any(
            isinstance(k, int) and not isinstance(k, bool) for k in key_consts
        )
        if has_int_key:
            report.add("warning", "FT009",
                       'minimal_roi 的键应是字符串分钟数（如 "0" / "30"），'
                       "不要写成整数 {0: ...}；Freqtrade 约定用字符串键",
                       line=lineno)
        # "0" 或 0 都算有零键（freqtrade 内部 int(key) 归一化）
        has_zero_key = any(
            (isinstance(k, str) and k.strip() == "0")
            or (isinstance(k, int) and not isinstance(k, bool) and k == 0)
            for k in key_consts
        )
        if value.keys and not has_zero_key:
            report.add("error", "FT008",
                       'minimal_roi 缺 "0" 键；没有它 Freqtrade 在 ROI 检查时'
                       "会 max() 空列表崩溃（见 freqtrade#1633）",
                       line=lineno)

    for item in strategy_class.body:
        if isinstance(item, ast.Assign):
            for target in item.targets:
                if isinstance(target, ast.Name) and target.id == "stoploss":
                    _check_stoploss(item.value, item.lineno)
                elif isinstance(target, ast.Name) and target.id == "minimal_roi":
                    _check_minimal_roi(item.value, item.lineno)
        elif isinstance(item, ast.AnnAssign) and isinstance(item.target, ast.Name):
            if item.value is None:
                continue
            if item.target.id == "stoploss":
                _check_stoploss(item.value, item.lineno)
            elif item.target.id == "minimal_roi":
                _check_minimal_roi(item.value, item.lineno)

    # 9. 检查必要 import：必须 from freqtrade.* 或 import freqtrade*
    has_freqtrade_import = False
    for stmt in ast.walk(tree):
        if isinstance(stmt, ast.ImportFrom):
            if stmt.module and stmt.module.startswith("freqtrade"):
                has_freqtrade_import = True
                break
        elif isinstance(stmt, ast.Import):
            for alias in stmt.names:
                if alias.name.startswith("freqtrade"):
                    has_freqtrade_import = True
                    break
            if has_freqtrade_import:
                break
    if not has_freqtrade_import:
        report.add("warning", "FT007",
                   "代码里看不到 from freqtrade.strategy import IStrategy")

    return report


def render_report(report: ValidationReport) -> str:
    """把报告渲染成人类可读字符串。"""
    lines = ["=" * 60, "Freqtrade Strategy Validation", "=" * 60]
    if report.passed and not report.warnings:
        lines.append("PASSED — 没有问题")
        return "\n".join(lines)
    lines.append(f"Errors: {len(report.errors)}    Warnings: {len(report.warnings)}")
    lines.append("")
    sev_order = {"error": 0, "warning": 1}
    for issue in sorted(report.issues, key=lambda i: (sev_order[i.severity], i.line)):
        tag = "[ERROR]" if issue.severity == "error" else "[WARN] "
        loc = f"L{issue.line}" if issue.line else "global"
        lines.append(f"{tag} {issue.code} {loc}: {issue.message}")
    lines.append("=" * 60)
    return "\n".join(lines)