# -*- coding: utf-8 -*-
"""
evolution/evolution_engine.py — 策略自我进化引擎（每周六 22:00，hermes cron）

设计依据：docs/decisions/2026-08-15-策略自我进化闭环重构.md（老莫已批准）
闭环：统计数据每日自动更新 → 本引擎每周检测退化 → 生成参数变体 → 回测验证 → 有价值才推送

流程：
  1. 读激活策略集合（data/strategy_weights.json：A股 active + 港股 markets.hk.active）
  2. 退化信号检测（宁缺毋滥，任一命中即触发研究）：
     S1 健康度连续低：strategy_health 连续 5 天 health_score < 40（排除 50 中性=无实盘数据）
     S2 温区表现衰减：激活策略在其适应温区（strategy_regime_perf）温区级组合年化 cagr_pct < 0
  3. 有退化 → 生成参数变体：
     - 只对 lab.STRATEGIES 里可回测的策略（v_oversold/v_weak 等标准回测体系）
     - 参数空间从策略 config 实际数值字段出发（递归遍历，单变量 ±20%，一次只动一个）
     - 港股走 hk_backtest（entry/exit 字段 ±20%）
  4. 回测验证（统一资金约束）：
     - A股：lab.run_backtest(save=False)，取 portfolio_full
     - 港股：hk_backtest.gen_trades_defensive + lab.portfolio_sim(max_positions=8)
     - 验收：温区级组合年化 cagr_pct ≥ 原策略 + 3pp 且 max_dd 不劣化超过 2pp
  5. 达标变体 → 写 strategy_evolution（promoted=0）+ XMPP 推送老莫（附对比证据）
     （永不自动 promote，老莫说"上线"才进路由）
  6. 无退化或变体全灭 → 当周静默（不制造噪音）

单例守卫：fcntl.flock 防并发（deploy_guard / 手动重跑均安全）
"""
import sys, os, json, sqlite3, copy, io, traceback
from datetime import datetime, timedelta

sys.path.insert(0, "/home/hmo/MoFin")
sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
sys.path.insert(0, "/home/hmo/MoFin/evolution")

DB = os.environ.get("MOFIN_DB", "/home/hmo/MoFin/data/mofin.db")
WEIGHTS_JSON = "/home/hmo/MoFin/data/strategy_weights.json"

# 退化信号参数
HEALTH_LOW = 40          # 健康度低于此值视为低
HEALTH_STREAK_DAYS = 5   # 连续天数
REGIME_CAGR_BAD = 0.0    # 温区级组合年化低于此值视为退化

# 变体生成参数
VAR_PCT = 0.20           # ±20% 网格
MAX_VARIANTS = 6         # 每策略最多生成变体数
MAX_VARIANTS_TEST = 1    # 最多回测验证的变体数（资源约束：单变体2y全市场回测6-8分钟/6-8GB，详见下方BT注释）

# 验证回测周期（2026-08-15：原5y全市场回测单变体8+分钟/5GB内存，改为2y控制资源；
# 验收对比用同周期原策略数据，相对改善仍有效）
BT_START = "2024-07-01"
BT_END = "2026-07-24"
BT_PERIOD_TAG = "2y"

# 验收门槛（2026-08-15 口径说明：变体与 parent 用同周期 strategy_research 2y 整体组合年化对比，
# 相对改善有效；原设计"温区级组合年化"需按温区分段重跑变体，资源过重，整体同口径更务实）
ACCEPT_CAGR_PP = 3.0     # 组合年化 ≥ 原 + 3pp
ACCEPT_DD_PP = 2.0       # max_dd 不劣化超过 2pp


def log(msg):
    line = f"[{datetime.now().isoformat(timespec='seconds')}] {msg}"
    print(line, flush=True)


# ── 单例守卫（fcntl，Windows 不可用则跳过）──
try:
    import fcntl
    _LOCK_FD = open("/tmp/evolution_engine.lock", "w")
    try:
        fcntl.flock(_LOCK_FD, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        log("已有 evolution_engine 实例在运行，退出")
        sys.exit(0)
except ImportError:
    pass


def get_conn():
    conn = sqlite3.connect(DB)
    conn.row_factory = sqlite3.Row
    return conn


# ── 1. 激活策略集合 ──
def load_active_strategies():
    """返回 [(version, market, regime)]：A股 active + 港股 markets.hk.active"""
    try:
        d = json.load(io.open(WEIGHTS_JSON, encoding="utf-8"))
    except Exception as e:
        log(f"读 strategy_weights.json 失败: {e}")
        return []
    out = []
    for v in (d.get("active") or []):
        info = (d.get("weights") or {}).get(v, {})
        out.append({"version": v, "market": "a",
                    "regime": info.get("best_regime") or info.get("regime") or d.get("state")})
    hk = (d.get("markets") or {}).get("hk") or {}
    for v in (hk.get("active") or []):
        out.append({"version": v, "market": "hk", "regime": hk.get("state")})
    return out


# ── 2. 退化信号检测 ──
def detect_degradation(conn, version, market):
    """返回退化原因列表（空=健康）。S1 健康度连续低；S2 温区组合年化<0"""
    reasons = []

    # S1：健康度连续 5 天 < 40（排除 50 中性=无实盘）
    rows = conn.execute(
        "SELECT date, health_score FROM strategy_health WHERE strategy_version=? ORDER BY date DESC LIMIT ?",
        (version, HEALTH_STREAK_DAYS)).fetchall()
    if len(rows) >= HEALTH_STREAK_DAYS:
        scores = [r["health_score"] for r in rows]
        # 排除"无实盘=50中性"污染：只要连续5天都 < 40 且不是 50 占位
        if all(s is not None and s < HEALTH_LOW for s in scores) and any(s != 50 for s in scores):
            reasons.append(f"S1 健康度连续{HEALTH_STREAK_DAYS}天<{HEALTH_LOW}（{scores}）")

    # S2：适应温区温区级组合年化 < 0（strategy_regime_perf.cagr_pct）
    r = conn.execute(
        "SELECT regime, cagr_pct, trades FROM strategy_regime_perf WHERE strategy=? AND market=? ORDER BY trades DESC LIMIT 1",
        (version, market)).fetchone()
    if r and r["cagr_pct"] is not None and r["cagr_pct"] < REGIME_CAGR_BAD:
        reasons.append(f"S2 适应温区[{r['regime']}]组合年化{r['cagr_pct']}%<0（{r['trades']}笔）")

    return reasons


# ── 3. 变体生成（数据驱动，从实际 config 出发）──
_NUM_KEYS = ("tp_pct", "sl_pct", "sl_atr", "max_hold_days", "min_score", "min_momentum",
             "adx_min", "atr_pct_min", "atr_pct_max", "roc_min", "roc_max",
             "macd_hist_min", "macd_hist_max", "dist_ma20_min", "vol_ratio_min",
             "vol_ratio_max", "ma20_slope_max", "mkt_slope_max", "mkt_adx_min",
             "sector_slope_max", "bias_max", "rsi_max", "ret_max", "mom20_max",
             "amount_max", "rsi_delta_min", "mkt_rsi_max", "mkt_dd60_max",
             "mcap_q_max", "pe_q_max", "news3_min", "sec_ret20_max",
             "pe_q_max", "mcap_q_max", "sec_ret20_min", "bias60_max",
             "vol_ratio_min", "rsi_delta_min", "bias60_min")
_SKIP_KEYS = ("mode", "family", "launch", "version", "name", "summary", "hypothesis",
              "mkt_mode", "mkt_above_ma20", "hh_only", "hl_only", "sector_above_ma20")


def iter_numeric_fields(node, path=()):
    """递归遍历 config，产出 (path_list, field_name, value) 数值字段"""
    if isinstance(node, dict):
        for k, v in node.items():
            if k in _SKIP_KEYS:
                continue
            if isinstance(v, (int, float)) and not isinstance(v, bool) and k in _NUM_KEYS:
                yield list(path) + [k], k, v
            elif isinstance(v, dict):
                yield from iter_numeric_fields(v, list(path) + [k])


def get_parent_cagr(conn, version, market, period_tag=BT_PERIOD_TAG):
    """原策略基准：优先同周期 strategy_research（period_tag=2y），无则温区级组合年化（全量）"""
    if market == "a":
        r = conn.execute(
            "SELECT results_json FROM strategy_research WHERE version=? AND period_tag=? "
            "AND market='a' ORDER BY id DESC LIMIT 1", (version, period_tag)).fetchone()
        if r:
            res = json.loads(r["results_json"] or "{}")
            s = res.get("summary", {})
            pf = s.get("portfolio_full", {})
            cagr = pf.get("cagr_pct")
            dd = pf.get("portfolio_max_dd_pct")
            if cagr is not None:
                return cagr, dd
    # 回退：温区级组合年化（strategy_regime_perf，全量）——仅当同周期数据缺失时
    r = conn.execute(
        "SELECT cagr_pct, portfolio_max_dd_pct FROM strategy_regime_perf WHERE strategy=? AND market=? ORDER BY trades DESC LIMIT 1",
        (version, market)).fetchone()
    if r:
        return r["cagr_pct"], r["portfolio_max_dd_pct"]
    return None, None


def generate_hypothesis_variants(version, market, base_config, period_tag=BT_PERIOD_TAG):
    """2026-08-16 数据归纳假设变体：从交易数据归纳可描述条件 → 生成加条件的策略版本
    假设格式：{feature, direction(max/min), threshold} → 对应入场条件
    返回 [{version, name, config, change_desc, evidence, hypothesis}]
    """
    try:
        from hypothesis_miner import induce_hypotheses
        hs, _ = induce_hypotheses(version, market, period_tag=period_tag)
    except Exception:
        hs = []
    variants = []
    for h in hs[:MAX_VARIANTS_TEST]:
        feat = h["feature"]
        direction = h["direction"]
        threshold = h["threshold"]
        # 映射到策略 config 的字段（A股 entry.filters/mr，港股 entry 顶层）
        cfg = copy.deepcopy(base_config)
        if market == "hk":
            entry = cfg.get("entry", {})
        else:
            entry = cfg.get("entry", {})
        # 字段名映射：面板字段 → 策略字段（多数同名，A股 mr 下）
        key = feat
        target = entry
        # A股 config 是 {entry:{filters,mr}} 结构，找可放的位置
        if market != "hk":
            if "mr" in entry:
                target = entry["mr"]
            elif "filters" in entry:
                target = entry["filters"]
        if direction == "max":
            target[key + "_max"] = threshold
        else:
            target[key + "_min"] = threshold
        vname = f"evo_{version}_{key}_{direction}{threshold}"
        variants.append({
            "version": vname,
            "name": f"自进化-{version}-规避{key}{direction}{threshold}",
            "config": cfg,
            "change_desc": f"[数据归纳] {h['hypothesis']}",
            "evidence": h.get("evidence", ""),
            "hypothesis": h.get("hypothesis", ""),
            "field": key,
            "delta": 0,
        })
    return variants


def generate_variants(version, market, config):
    """生成变体参数建议：单变量 ±20%，最多 MAX_VARIANTS 个
    返回 [{version, name, config, change_desc, field, delta}]"""
    fields = list(iter_numeric_fields(config))
    if not fields:
        return []
    variants = []
    for path, fname, val in fields:
        if val <= 0:
            continue
        for factor, tag in [(1 - VAR_PCT, "减20%"), (1 + VAR_PCT, "加20%")]:
            new_val = round(val * factor, 4)
            if new_val <= 0:
                continue
            # 克隆 config 并修改目标字段
            new_cfg = copy.deepcopy(config)
            node = new_cfg
            for p in path[:-1]:
                node = node[p]
            node[path[-1]] = new_val
            variants.append({
                "version": f"evo_{version}_{fname}_{tag.replace('20%','')}{round(new_val, 2)}",
                "name": f"自进化-{version}-{fname}{tag}",
                "config": new_cfg,
                "change_desc": f"{fname}: {val} → {new_val}（{tag}）",
                "field": fname,
                "delta": round(new_val - val, 4),
            })
            if len(variants) >= MAX_VARIANTS:
                return variants
    return variants


# ── 4. 回测验证 ──
def verify_variant_a(variant, parent_version):
    """A股变体验证：注册进 lab 跑回测（save=False），返回 summary 关键指标"""
    import strategy_lab as lab
    name = variant["version"]
    base = lab.get_strategy(parent_version)
    cfg = copy.deepcopy(base)
    cfg["version"] = name
    cfg["name"] = variant["name"]
    # 用变体 config 覆盖（变体 config 从原 config 克隆并改了一个字段）
    merged = copy.deepcopy(base["config"])
    _deep_update(merged, variant["config"])
    cfg["config"] = merged
    lab.STRATEGIES[name] = cfg
    try:
        r = lab.run_backtest(name, BT_START, BT_END, 913000, save=False,
                             universe="a", period_tag=BT_PERIOD_TAG)
        s = r.get("summary", {})
        pf = s.get("portfolio_full", {})
        return {
            "trades": s.get("total_trades"),
            "win_rate": s.get("win_rate"),
            "cagr": pf.get("cagr_pct"),
            "total_return": pf.get("total_return_pct"),
            "max_dd": pf.get("portfolio_max_dd_pct"),
        }
    finally:
        lab.STRATEGIES.pop(name, None)


def _deep_update(dst, src):
    for k, v in src.items():
        if isinstance(v, dict) and isinstance(dst.get(k), dict):
            _deep_update(dst[k], v)
        else:
            dst[k] = v


def verify_variant_hk(variant, parent_version):
    """港股变体验证：hk_backtest 生成交易 + portfolio_sim 8槽"""
    import pandas as pd
    sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
    from hk_strategies import HK_STRATEGIES, get_hk_strategy
    import hk_backtest
    import strategy_lab as lab

    base = get_hk_strategy(parent_version)
    if not base:
        return None
    new_cfg = copy.deepcopy(base)
    new_cfg["version"] = variant["version"]
    new_cfg["name"] = variant["name"]
    _deep_update(new_cfg, variant["config"])
    panel = hk_backtest.load_panel()
    # 2y 窗口过滤（与 A股验证周期一致，控制资源）
    panel = panel[(panel["date"] >= BT_START) & (panel["date"] <= BT_END)].copy()
    trades = hk_backtest.gen_trades_defensive(panel, new_cfg, strike=3, cooldown_days=15)
    if not trades:
        return {"trades": 0, "win_rate": None, "cagr": None, "total_return": None, "max_dd": None}
    sim = lab.portfolio_sim(trades, 1000000, max_positions=8)
    return {
        "trades": len(trades),
        "win_rate": round(100 * sum(1 for t in trades if t["profit_pct"] > 0) / len(trades), 1),
        "cagr": sim.get("cagr_pct"),
        "total_return": sim.get("total_return_pct"),
        "max_dd": sim.get("portfolio_max_dd_pct"),
    }


# ── 5. 记录 + 推送 ──
def record_and_notify(conn, parent_version, market, variant, result, parent_cagr, parent_dd):
    """写 strategy_evolution + XMPP 推送"""
    conn.execute("""
        INSERT INTO strategy_evolution (parent_version, child_version, change_description, backtest_result, promoted)
        VALUES (?, ?, ?, ?, 0)
    """, (parent_version, variant["version"], variant["change_desc"],
          json.dumps(result, ensure_ascii=False)))
    conn.commit()

    msg = (f"🧬 策略进化建议 [{parent_version}]\n"
           f"改动: {variant['change_desc']}\n"
           f"回测: 年化 {parent_cagr}% → {result.get('cagr')}%"
           f" (Δ{round((result.get('cagr') or 0) - (parent_cagr or 0), 1)}pp)"
           f" | 回撤 {parent_dd}% → {result.get('max_dd')}%\n"
           f"胜率 {result.get('win_rate')}% / {result.get('trades')}笔\n"
           f"【验证达标，待你决定是否上线】")
    try:
        sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
        from alert_helper import notify, ACTION
        notify("策略进化", msg, level=ACTION)
        log(f"XMPP 推送: {parent_version} → {variant['version']}")
    except Exception as e:
        log(f"XMPP 推送失败: {e}")
    return msg


# ── 主流程 ──
def run_evolution():
    conn = get_conn()
    actives = load_active_strategies()
    log(f"激活策略: {[a['version'] for a in actives]}")
    if not actives:
        log("无激活策略，退出")
        conn.close()
        return

    findings = []   # 退化发现
    passed = []     # 达标变体

    for act in actives:
        v, mkt = act["version"], act["market"]
        reasons = detect_degradation(conn, v, mkt)
        if not reasons:
            continue
        log(f"退化信号: {v} [{mkt}] → {'; '.join(reasons)}")
        findings.append((v, mkt, reasons))

        # 生成变体（A股从 lab 读 config；港股从 HK_STRATEGIES）
        if mkt == "hk":
            sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
            from hk_strategies import get_hk_strategy
            base = get_hk_strategy(v)
            if not base:
                log(f"  {v} 无港股策略定义，跳过")
                continue
            # 2026-08-16 优先数据归纳假设，无则参数变体
            variants = generate_hypothesis_variants(v, mkt, base)
            if not variants:
                variants = generate_variants(v, mkt, base)
            verify_fn = verify_variant_hk
        else:
            try:
                import strategy_lab as lab
                base = lab.get_strategy(v)
            except ValueError:
                log(f"  {v} 不在标准回测体系（scanner 类策略），跳过变体研究")
                continue
            # 2026-08-16 优先数据归纳假设，无则参数变体
            variants = generate_hypothesis_variants(v, mkt, base["config"])
            if not variants:
                variants = generate_variants(v, mkt, base["config"])
            verify_fn = verify_variant_a
        if not variants:
            log(f"  {v} 无可用变体字段，跳过")
            continue

        parent_cagr, parent_dd = get_parent_cagr(conn, v, mkt)
        log(f"  {v} 原温区年化 {parent_cagr}% / 回撤 {parent_dd}% | 生成 {len(variants)} 个变体，验证前 {MAX_VARIANTS_TEST} 个")
        for var in variants[:MAX_VARIANTS_TEST]:
            try:
                res = verify_fn(var, v)
            except Exception as e:
                log(f"  {var['version']} 回测失败: {str(e)[:100]}")
                continue
            if not res or res.get("cagr") is None:
                log(f"  {var['version']} 无结果（0笔或空），跳过")
                continue
            ok_cagr = parent_cagr is None or res["cagr"] >= (parent_cagr or 0) + ACCEPT_CAGR_PP
            ok_dd = parent_dd is None or res["max_dd"] <= (parent_dd or 0) + ACCEPT_DD_PP
            status = "✅达标" if (ok_cagr and ok_dd) else "❌不达标"
            log(f"  {var['version']}: 年化 {parent_cagr}→{res['cagr']}% 回撤 {parent_dd}→{res['max_dd']}% [{status}]")
            if ok_cagr and ok_dd:
                record_and_notify(conn, v, mkt, var, res, parent_cagr, parent_dd)
                passed.append((v, var, res))

    conn.close()

    # 汇总
    if not findings:
        log("── 无退化信号，当周静默 ──")
    else:
        log(f"── 检测 {len(findings)} 个退化策略，{len(passed)} 个达标变体已推送 ──")
    return findings, passed


if __name__ == "__main__":
    try:
        run_evolution()
    except Exception as e:
        log(f"evolution_engine 异常: {e}")
        traceback.print_exc()
        sys.exit(1)