单品种 AI 自进化交易系统

# -*- coding: utf-8 -*-
"""
单品种 AI 自进化交易系统 - 发明者量化(FMZ.COM) Python V1
适用标的：BTC / 黄金 / 原油 等任意单一 TradFi 或加密货币永续合约

=====================================================================
设计核心：Decide -> Execute -> Reflect -> Evolve 闭环自进化
=====================================================================
1. 指标层：纯 Python 手写多组技术指标(趋势/动量/波动率/量能)，不依赖
   平台 TA 库具体版本签名，保证跨环境稳定运行。
2. 新闻层：BraveSearch 抓取该品种相关新闻，去重缓存进数据库。
3. 决策层：LLM 综合"指标快照 + 新闻 + 当前 Playbook(经验库)"输出结构化
   交易信号，要求技术面与消息面不冲突才允许开仓/加仓。
4. 执行层：仓位/止损/移动止盈参数从 Playbook 读取(不是写死常量)，
   支持同向加仓、反向平仓再反手，内置硬止损与移动止盈。
5. 复盘层：每笔平仓后，LLM 分析这笔交易的输赢原因，提炼中文经验写入
   数据库(reflections 表)。
6. 进化层：每积累 N 笔复盘(或每隔固定时间)，LLM 读取近期复盘 + 整体
   胜率统计，输出新版本 Playbook(新的 do/avoid 规则 + 参数建议)。
   参数建议会被裁剪进人工设定的安全上下限，写回数据库后立即生效，
   下一次决策自动使用最新 Playbook —— 这就是"自进化"的闭环。

安全设计：
- AI 不能修改代码，只能在人工设定的硬性范围内微调数值参数。
- 硬止损、单标的仓位上限、最大加仓次数是代码层强约束，AI 建议无法突破。
- 开仓强制要求"技术面 + 消息面"双重不冲突确认。
- 默认 notify 模式，仅通知不实盘。

=====================================================================
以下常量由平台参数注入(在 FMZ 策略"参数"面板中创建，无需在代码中硬编码)：

  LLM_API_KEY / LLM_BASE_URL / LLM_MODEL      LLM 接口配置
  BRAVE_KEY                                    BraveSearch Key(留空跳过新闻)
  TRADE_MODE                                   notify 仅通知 / trade 实盘

  INSTRUMENT_KEYWORD                           合约匹配关键字，如 "BTC_USDT" "XAU_USDT" "WTI_USDT"
  INSTRUMENT_DISPLAY_NAME                      中文显示名，如 "比特币" "黄金" "WTI原油"
  NEWS_QUERY_TERMS                             逗号分隔的新闻检索关键词，如 "美联储,ETF,减半"
  KLINE_PERIOD_LABEL                           M15 / M30 / H1 / H4 / D1

  CHECK_INTERVAL_S                             主循环间隔秒
  NEWS_REFRESH_S                                新闻刷新间隔秒
  DECISION_COOLDOWN_S                          两次AI决策最小间隔秒
  KLINE_LIMIT                                  拉取K线根数

  MIN_CONFIDENCE / MIN_CONFIDENCE_FLOOR / MIN_CONFIDENCE_CEIL       最小置信度(初始值/进化下限/进化上限)
  BASE_POS_PCT / BASE_POS_PCT_FLOOR / BASE_POS_PCT_CEIL             基础仓位比例(初始值/进化下限/进化上限)
  MAX_SINGLE_POS_PCT                           单标的最大仓位比例(硬约束，不参与进化)
  MAX_TOTAL_ADD_COUNT                          最大加仓次数(硬约束)
  LEVERAGE                                     杠杆(硬约束)

  HARD_STOP_PCT / HARD_STOP_PCT_MIN / HARD_STOP_PCT_MAX             硬止损%(初始值/进化下限/进化上限)
  TRAIL_ACTIVATE_PCT / TRAIL_ACTIVATE_PCT_MIN / TRAIL_ACTIVATE_PCT_MAX   移动止盈激活%
  TRAIL_GIVEBACK_PCT / TRAIL_GIVEBACK_PCT_MIN / TRAIL_GIVEBACK_PCT_MAX   移动止盈回撤%

  EVOLUTION_TRADE_INTERVAL                     每积累几笔已复盘交易触发一次进化
  EVOLUTION_TIME_INTERVAL_S                    兜底：每隔多久强制检查一次进化
  EVOLUTION_MIN_TRADES                         至少要有几笔复盘记录才允许第一次进化
  REFLECTION_LOOKBACK                          进化时回顾最近几笔复盘记录
=====================================================================
"""

import json
import math
import re
import time
import urllib.parse
import urllib.request


STORE_KEY = "ai_single_evolve_store_v1"
POSITION_STATE_KEY = "ai_single_evolve_pos_v1"


# =========================
# 主循环
# =========================

def main():
    Log("单品种AI自进化交易系统启动 | {} | mode:{}".format(INSTRUMENT_DISPLAY_NAME, TRADE_MODE), "#00AAFF")
    db_init()
    store = load_store()
    if not store.get("market"):
        resolve_market(store)
    load_or_init_playbook(store)
    last_dashboard = 0

    while True:
        try:
            handle_command(store)

            if not store.get("market"):
                resolve_market(store)

            now = int(time.time())

            if store.get("market"):
                if BRAVE_KEY and now - store.get("last_news_refresh", 0) > NEWS_REFRESH_S:
                    news = fetch_instrument_news(store)
                    store["news_cache_recent"] = news
                    store["last_news_refresh"] = now
                    save_store(store)

                if now - store.get("last_decision_time", 0) > DECISION_COOLDOWN_S:
                    run_decision_cycle(store)
                    store["last_decision_time"] = now
                    save_store(store)

                monitor_position(store)
                maybe_run_evolution(store)

            if now - last_dashboard > 60:
                show_dashboard(store)
                last_dashboard = now

            save_store(store)

        except Exception as e:
            Log("主循环异常: " + str(e), "#FF0000")

        Sleep(CHECK_INTERVAL_S * 1000)


# =========================
# Store (工作记忆，轻量易变状态)
# =========================

def new_store():
    return {
        "market": None,
        "indicator_snapshot": {},
        "news_cache_recent": [],
        "current_playbook": {},
        "signals": [],
        "step_logs": [],
        "closed_trade_count_since_evolution": 0,
        "evolution_paused": False,
        "last_news_refresh": 0,
        "last_decision_time": 0,
        "last_evolution_check": 0,
    }


def load_store():
    store = _G(STORE_KEY)
    if not store:
        return new_store()
    fresh = new_store()
    for k, v in fresh.items():
        if k not in store:
            store[k] = v
    return store


def save_store(store):
    _G(STORE_KEY, store)


def record_step(store, step, data, color="#999999"):
    row = {"time": int(time.time()), "step": step, "data": data}
    store.setdefault("step_logs", []).insert(0, row)
    store["step_logs"] = store["step_logs"][:300]
    Log("STEP | {} | {}".format(step, json.dumps(data, ensure_ascii=False)[:400]), color)


def handle_command(store):
    cmd = GetCommand()
    if not cmd:
        return
    if cmd == "decision:force":
        store["last_decision_time"] = 0
        save_store(store)
        Log("已强制触发下一轮决策", "#00CC66")
    elif cmd == "evolution:trigger":
        run_evolution(store, forced=True)
    elif cmd == "evolution:pause":
        store["evolution_paused"] = True
        save_store(store)
        Log("进化已暂停", "#FFAA00")
    elif cmd == "evolution:resume":
        store["evolution_paused"] = False
        save_store(store)
        Log("进化已恢复", "#00CC66")
    elif cmd == "clear:store":
        market = store.get("market")
        store.clear()
        store.update(new_store())
        store["market"] = market
        load_or_init_playbook(store)
        save_store(store)
        Log("运行态已重置(数据库历史交易/经验保留)", "#FFAA00")
    elif cmd == "db:wipe":
        for tbl in ["indicator_snapshot", "decisions", "trades", "reflections", "playbook", "news_cache"]:
            db_exec("DELETE FROM {}".format(tbl))
        store.clear()
        store.update(new_store())
        resolve_market(store)
        load_or_init_playbook(store)
        save_store(store)
        Log("数据库与运行态已全部清空", "#FF0000")


# =========================
# 数据库层 (DBExec / SQLite - 长期记忆)
# =========================

def db_init():
    db_exec('CREATE TABLE IF NOT EXISTS indicator_snapshot ('
            'id INTEGER PRIMARY KEY AUTOINCREMENT, time INTEGER NOT NULL, '
            'price REAL, snapshot_json TEXT NOT NULL);')
    db_exec('CREATE TABLE IF NOT EXISTS decisions ('
            'id INTEGER PRIMARY KEY AUTOINCREMENT, time INTEGER NOT NULL, '
            'playbook_version INTEGER, action TEXT, direction TEXT, confidence INTEGER, '
            'technical_view TEXT, news_view TEXT, reason TEXT, '
            'indicators_json TEXT, news_json TEXT, executed INTEGER DEFAULT 0);')
    db_exec('CREATE TABLE IF NOT EXISTS trades ('
            'id INTEGER PRIMARY KEY AUTOINCREMENT, open_time INTEGER, close_time INTEGER, '
            'direction TEXT, entry_price REAL, exit_price REAL, qty REAL, pnl_pct REAL, '
            'close_reason TEXT, decision_id INTEGER, reflected INTEGER DEFAULT 0);')
    db_exec('CREATE TABLE IF NOT EXISTS reflections ('
            'id INTEGER PRIMARY KEY AUTOINCREMENT, trade_id INTEGER, time INTEGER, '
            'outcome TEXT, mistake_type TEXT, lesson TEXT, tags TEXT, raw_json TEXT);')
    db_exec('CREATE TABLE IF NOT EXISTS playbook ('
            'id INTEGER PRIMARY KEY AUTOINCREMENT, version INTEGER, created_at INTEGER, '
            'summary TEXT, do_rules TEXT, avoid_rules TEXT, params_json TEXT, based_on_trades INTEGER);')
    db_exec('CREATE TABLE IF NOT EXISTS news_cache ('
            'id INTEGER PRIMARY KEY AUTOINCREMENT, time INTEGER, url TEXT UNIQUE, '
            'title TEXT, desc TEXT, source TEXT, query TEXT);')


def db_exec(sql, *params):
    try:
        return DBExec(sql, *params) if params else DBExec(sql)
    except Exception as e:
        Log("DB执行失败 | {} | {}".format(sql[:100], str(e)), "#FF0000")
        return None


def db_query(sql, *params):
    res = db_exec(sql, *params)
    cols, vals = _extract_cols_values(res)
    out = []
    for row in vals:
        out.append({cols[i]: row[i] for i in range(min(len(cols), len(row)))})
    return out


def _extract_cols_values(res):
    if res is None:
        return [], []
    if isinstance(res, dict):
        return res.get("columns", []) or [], res.get("values", []) or []
    return getattr(res, "columns", []) or [], getattr(res, "values", []) or []


def db_last_id():
    rows = db_query("select last_insert_rowid() as id")
    return rows[0]["id"] if rows else None


# =========================
# 品种解析
# =========================

def resolve_market(store):
    try:
        ms = exchange.GetMarkets()
    except Exception as e:
        Log("GetMarkets失败: " + str(e), "#FF0000")
        return
    found_key, found_market = None, None
    kw = INSTRUMENT_KEYWORD.upper()
    # 优先级：永续 swap > 无后缀精确匹配 > 子串回退。
    # 避免 "BTC_USDT" 误命中 "BTC_USDT.next_quarter" 等季度合约。
    swap_hit = exact_hit = sub_hit = None
    for key, market in ms.items():
        ku = key.upper()
        if ku == kw + ".SWAP":
            swap_hit = (key, market)
            break
        if ku == kw:
            exact_hit = exact_hit or (key, market)
        elif kw in ku and sub_hit is None:
            sub_hit = (key, market)
    picked = swap_hit or exact_hit or sub_hit
    if picked:
        found_key, found_market = picked
    if not found_key:
        Log("未找到匹配 {} 的合约，请检查 INSTRUMENT_KEYWORD 参数".format(INSTRUMENT_KEYWORD), "#FF0000")
        return
    store["market"] = {
        "contract": found_key,
        "amountPrecision": found_market.get("AmountPrecision", 0),
        "pricePrecision": found_market.get("PricePrecision", 2),
        "ctVal": found_market.get("CtVal", 1) or 1,
        "minQty": found_market.get("MinQty", 0) or 0,
    }
    save_store(store)
    record_step(store, "market_resolved", {"contract": found_key}, "#00AAFF")


def resolve_period():
    names = {"M15": "PERIOD_M15", "M30": "PERIOD_M30", "H1": "PERIOD_H1",
             "H4": "PERIOD_H4", "D1": "PERIOD_D1"}
    label = str(KLINE_PERIOD_LABEL).upper() if "KLINE_PERIOD_LABEL" in globals() else "H1"
    name = names.get(label, "PERIOD_H1")
    return globals().get(name, globals().get("PERIOD_H1"))


KLINE_PERIOD = None  # 在 main 首次使用前惰性解析，避免参数注入时序问题


def get_kline_period():
    global KLINE_PERIOD
    if KLINE_PERIOD is None:
        KLINE_PERIOD = resolve_period()
    return KLINE_PERIOD


def get_records(contract, limit):
    try:
        bars = exchange.GetRecords(contract, get_kline_period()) or []
        return bars[-limit:]
    except Exception as e:
        Log("GetRecords失败 | {} | {}".format(contract, str(e)), "#FFAA00")
        return []


# =========================
# 指标层 (纯 Python 手写，不依赖平台TA库签名)
# =========================

def ema_full_series(values, period):
    if not values:
        return []
    k = 2.0 / (period + 1)
    out = [values[0]]
    for v in values[1:]:
        out.append(out[-1] * (1 - k) + v * k)
    return out


def latest_ema(values, period):
    series = ema_full_series(values, period)
    return series[-1] if series else 0


def latest_rsi(values, period=14):
    if len(values) < period + 1:
        return 50.0
    deltas = [values[i] - values[i - 1] for i in range(1, len(values))]
    gains = [max(d, 0) for d in deltas]
    losses = [max(-d, 0) for d in deltas]
    avg_gain = mean(gains[:period])
    avg_loss = mean(losses[:period])
    for i in range(period, len(deltas)):
        avg_gain = (avg_gain * (period - 1) + gains[i]) / period
        avg_loss = (avg_loss * (period - 1) + losses[i]) / period
    if avg_loss == 0:
        return 100.0
    rs = avg_gain / avg_loss
    return 100 - 100 / (1 + rs)


def latest_macd(values, fast=12, slow=26, signal=9):
    if len(values) < slow + signal:
        return 0.0, 0.0, 0.0, 0.0
    fast_series = ema_full_series(values, fast)
    slow_series = ema_full_series(values, slow)
    dif_series = [f - s for f, s in zip(fast_series, slow_series)]
    dea_series = ema_full_series(dif_series, signal)
    dif = dif_series[-1]
    dea = dea_series[-1]
    hist = dif - dea
    hist_prev = (dif_series[-2] - dea_series[-2]) if len(dif_series) > 1 else hist
    return dif, dea, hist, hist_prev


def latest_boll(values, period=20, mult=2):
    window = values[-period:] if len(values) >= period else values
    mid = mean(window)
    sd = std(window)
    return mid + mult * sd, mid, mid - mult * sd


def latest_atr(records, period=14):
    if len(records) < 2:
        return 0.0
    trs = []
    for i in range(1, len(records)):
        h, l, pc = records[i]["High"], records[i]["Low"], records[i - 1]["Close"]
        trs.append(max(h - l, abs(h - pc), abs(l - pc)))
    window = trs[-period:] if len(trs) >= period else trs
    return mean(window) if window else 0.0


def latest_kdj(records, n=9, k_period=3, d_period=3):
    if len(records) < n:
        return 50.0, 50.0, 50.0
    closes = [r["Close"] for r in records]
    highs = [r["High"] for r in records]
    lows = [r["Low"] for r in records]
    k, d = 50.0, 50.0
    for i in range(n - 1, len(records)):
        hh = max(highs[i - n + 1:i + 1])
        ll = min(lows[i - n + 1:i + 1])
        rsv = (closes[i] - ll) / (hh - ll) * 100 if hh > ll else 50.0
        k = (k_period - 1) / k_period * k + 1 / k_period * rsv
        d = (d_period - 1) / d_period * d + 1 / d_period * k
    j = 3 * k - 2 * d
    return k, d, j


def pct_change(closes, n):
    if len(closes) <= n:
        return 0.0
    a, b = closes[-1 - n], closes[-1]
    return (b - a) / a * 100 if a else 0.0


def build_indicator_snapshot(records):
    if not records or len(records) < 60:
        return None
    closes = [r["Close"] for r in records]
    price = closes[-1]

    ema_fast = latest_ema(closes, 20)
    ema_slow = latest_ema(closes, 60)
    dif, dea, hist, hist_prev = latest_macd(closes, 12, 26, 9)
    rsi_v = latest_rsi(closes, 14)
    boll_upper, boll_mid, boll_lower = latest_boll(closes, 20, 2)
    atr_v = latest_atr(records, 14)
    k_v, d_v, j_v = latest_kdj(records, 9, 3, 3)

    vols = [r.get("Volume", 0) or 0 for r in records[-20:]]
    vol_ma20 = mean(vols) if vols else 0
    vol_now = records[-1].get("Volume", 0) or 0
    vol_ratio = vol_now / vol_ma20 if vol_ma20 > 0 else 1.0

    band_width = boll_upper - boll_lower
    band_pos = (price - boll_lower) / band_width if band_width > 0 else 0.5

    return {
        "price": round(price, 6),
        "trend": {
            "ema_fast": round(ema_fast, 6),
            "ema_slow": round(ema_slow, 6),
            "ema_bias_pct": round((ema_fast - ema_slow) / ema_slow * 100, 3) if ema_slow else 0,
            "macd_dif": round(dif, 6),
            "macd_dea": round(dea, 6),
            "macd_hist": round(hist, 6),
            "macd_hist_rising": bool(hist > hist_prev),
        },
        "momentum": {
            "rsi14": round(rsi_v, 2),
            "kdj_k": round(k_v, 2),
            "kdj_d": round(d_v, 2),
            "kdj_j": round(j_v, 2),
        },
        "volatility": {
            "boll_upper": round(boll_upper, 6),
            "boll_mid": round(boll_mid, 6),
            "boll_lower": round(boll_lower, 6),
            "band_position": round(clamp(band_pos, 0, 1), 3),
            "atr": round(atr_v, 6),
            "atr_pct": round(atr_v / price * 100, 3) if price else 0,
        },
        "volume": {"vol_ratio_to_ma20": round(vol_ratio, 3)},
        "price_action": {
            "pct_change_1bar": round(pct_change(closes, 1), 3),
            "pct_change_6bar": round(pct_change(closes, 6), 3),
            "pct_change_24bar": round(pct_change(closes, 24), 3),
        },
        "time": int(time.time()),
    }


# =========================
# 新闻层 (BraveSearch)
# =========================

def fetch_instrument_news(store):
    if not BRAVE_KEY:
        return []
    terms = [t.strip() for t in str(NEWS_QUERY_TERMS).split(",") if t.strip()]
    queries = [
        '"{}" 最新消息 价格走势'.format(INSTRUMENT_DISPLAY_NAME),
        '"{}" 行情分析'.format(INSTRUMENT_DISPLAY_NAME),
    ]
    for t in terms[:4]:
        queries.append('"{}" {}'.format(INSTRUMENT_DISPLAY_NAME, t))
    news = brave_news_search(queries[:6], 4)
    for item in news:
        db_exec('INSERT OR IGNORE INTO news_cache(time, url, title, desc, source, query) values(?,?,?,?,?,?)',
                int(time.time()), item.get("url", ""), item.get("title", ""),
                item.get("desc", ""), item.get("source", ""), item.get("query", ""))
    record_step(store, "news_refreshed", {"count": len(news)}, "#00AAFF")
    return news


def brave_news_search(queries, count):
    out, seen = [], {}
    for q in queries:
        try:
            url = ("https://api.search.brave.com/res/v1/news/search?q=" +
                   urllib.parse.quote(q) + "&count=" + str(count) + "&freshness=pd")
            raw = http_request(url, "GET", {"X-Subscription-Token": BRAVE_KEY, "Accept": "application/json"}, None, 20)
            data = json.loads(raw)
            for item in data.get("results", []):
                link = item.get("url", "") or item.get("title", "")
                if not link or link in seen:
                    continue
                seen[link] = True
                source = item.get("profile", {}) or {}
                out.append({
                    "title": item.get("title", "") or "",
                    "desc": item.get("description", "") or "",
                    "url": item.get("url", "") or "",
                    "age": item.get("age", "") or "",
                    "source": source.get("name", "") or "",
                    "query": q,
                })
        except Exception as e:
            Log("BraveSearch失败 | {} | {}".format(q, str(e)), "#FFAA00")
        Sleep(800)
    return out


# =========================
# Playbook 经验库(进化载体)
# =========================

def default_playbook_params():
    return {
        "min_confidence": MIN_CONFIDENCE,
        "base_pos_pct": BASE_POS_PCT,
        "hard_stop_pct": HARD_STOP_PCT,
        "trail_activate_pct": TRAIL_ACTIVATE_PCT,
        "trail_giveback_pct": TRAIL_GIVEBACK_PCT,
    }


def load_or_init_playbook(store):
    rows = db_query("select * from playbook order by version desc limit 1")
    if rows:
        row = rows[0]
        store["current_playbook"] = {
            "version": row.get("version", 1),
            "summary": row.get("summary", ""),
            "do_rules": safe_json_loads(row.get("do_rules"), []),
            "avoid_rules": safe_json_loads(row.get("avoid_rules"), []),
            "params": safe_json_loads(row.get("params_json"), default_playbook_params()),
            "based_on_trades": row.get("based_on_trades", 0),
        }
    else:
        params = default_playbook_params()
        now = int(time.time())
        db_exec('INSERT INTO playbook(version, created_at, summary, do_rules, avoid_rules, params_json, based_on_trades) '
                'values(?,?,?,?,?,?,?)',
                1, now, "初始版本，尚无交易经验，按基础参数运行。",
                json.dumps([], ensure_ascii=False), json.dumps([], ensure_ascii=False),
                json.dumps(params, ensure_ascii=False), 0)
        store["current_playbook"] = {
            "version": 1, "summary": "初始版本，尚无交易经验，按基础参数运行。",
            "do_rules": [], "avoid_rules": [], "params": params, "based_on_trades": 0,
        }
    save_store(store)


def playbook_param(store, key, config_default, lo, hi):
    v = store.get("current_playbook", {}).get("params", {}).get(key, config_default)
    try:
        v = float(v)
    except Exception:
        v = config_default
    return clamp(v, lo, hi)


# =========================
# 决策层
# =========================

def run_decision_cycle(store):
    market = store.get("market")
    if not market:
        return
    records = get_records(market["contract"], KLINE_LIMIT)
    if not records or len(records) < 60:
        return

    decision = make_decision(store, records)
    if not decision:
        return

    ticker = None
    try:
        ticker = exchange.GetTicker(market["contract"])
    except Exception as e:
        Log("GetTicker失败: " + str(e), "#FFAA00")
        return

    apply_decision(store, decision, ticker)


def make_decision(store, records):
    snapshot = build_indicator_snapshot(records)
    if not snapshot:
        return None
    store["indicator_snapshot"] = snapshot
    db_exec('INSERT INTO indicator_snapshot(time, price, snapshot_json) values(?,?,?)',
            snapshot["time"], snapshot["price"], json.dumps(snapshot, ensure_ascii=False))

    news = store.get("news_cache_recent", [])
    playbook = store.get("current_playbook", {})
    pos = _G(POSITION_STATE_KEY) or {}

    if not LLM_API_KEY:
        record_step(store, "decision_skipped_no_llm", {}, "#FFAA00")
        return None

    schema = {
        "action": "open_long|open_short|add_long|add_short|reduce|close|hold",
        "confidence": 0,
        "technical_view": "bullish|bearish|neutral|conflicted",
        "news_view": "bullish|bearish|neutral|no_data",
        "reason": "",
        "key_indicators_cited": [""],
        "suggested_stop_pct": 0.0,
        "risk_notes": "",
    }

    prompt = "\n".join([
        "你是{}单品种AI交易决策引擎。".format(INSTRUMENT_DISPLAY_NAME),
        "当前playbook经验(do_rules必须遵守，avoid_rules必须避免):",
        json.dumps(playbook, ensure_ascii=False),
        "当前持仓状态(空字典表示无持仓):",
        json.dumps(pos, ensure_ascii=False),
        "多指标快照:",
        json.dumps(snapshot, ensure_ascii=False),
        "近期相关新闻(可能为空):",
        json.dumps(news[:10], ensure_ascii=False),
        "输出schema:",
        json.dumps(schema, ensure_ascii=False),
        "要求：",
        "0. 综合趋势/动量/波动率/量能多个维度，不要只依赖单一指标下结论。",
        "1. 只有technical_view与news_view不冲突(news_view为neutral/no_data，或方向与technical_view一致)才能给出open_long/open_short/add_long/add_short。",
        "2. 如果news_view方向与technical_view相反，action必须是hold或close。",
        "3. 已有持仓且趋势明显反转时，action应为close，不要硬扛。",
        "4. confidence必须真实反映把握程度(0-100)，不要习惯性给高分，把握不足给低分。",
        "5. reason和risk_notes必须使用中文，并结合playbook中的do_rules/avoid_rules说明理由。",
        "6. 字段名必须和schema完全一致，不要新增或改名字段。",
    ])

    data = call_llm_json("你是严谨的量化交易决策引擎，只输出JSON，所有中文字段必须使用中文，不要输出markdown。", prompt, 1200)
    if not data:
        return None

    decision = sanitize_decision(data)
    db_exec('INSERT INTO decisions(time, playbook_version, action, direction, confidence, technical_view, news_view, '
            'reason, indicators_json, news_json, executed) values(?,?,?,?,?,?,?,?,?,?,0)',
            int(time.time()), playbook.get("version", 0), decision["action"], decision["direction"],
            decision["confidence"], decision["technical_view"], decision["news_view"], decision["reason"],
            json.dumps(snapshot, ensure_ascii=False), json.dumps(news[:10], ensure_ascii=False))
    decision["id"] = db_last_id()

    store.setdefault("signals", []).insert(0, {
        "time": int(time.time()), "action": decision["action"], "confidence": decision["confidence"],
        "technical_view": decision["technical_view"], "news_view": decision["news_view"],
        "reason": decision["reason"],
    })
    store["signals"] = store["signals"][:50]

    record_step(store, "decision_made", {
        "action": decision["action"], "confidence": decision["confidence"],
        "technical_view": decision["technical_view"], "news_view": decision["news_view"],
    }, "#00AAFF")
    return decision


def sanitize_decision(data):
    action = str(data.get("action", "hold") or "hold").lower()
    valid_actions = ["open_long", "open_short", "add_long", "add_short", "reduce", "close", "hold"]
    if action not in valid_actions:
        action = "hold"
    direction = "long" if "long" in action else ("short" if "short" in action else "")
    return {
        "action": action,
        "direction": direction,
        "confidence": normalize_confidence(data.get("confidence", 0)),
        "technical_view": data.get("technical_view", "neutral") or "neutral",
        "news_view": data.get("news_view", "no_data") or "no_data",
        "reason": data.get("reason", "") or "",
        "key_indicators_cited": as_list(data.get("key_indicators_cited")),
        "suggested_stop_pct": safe_float(data.get("suggested_stop_pct", 0)),
        "risk_notes": data.get("risk_notes", "") or "",
    }


# =========================
# 执行层
# =========================

def apply_decision(store, decision, ticker):
    if not ticker:
        return
    action = decision["action"]
    pos = _G(POSITION_STATE_KEY) or {}
    has_pos = bool(pos and pos.get("qty", 0) > 0)
    min_conf = playbook_param(store, "min_confidence", MIN_CONFIDENCE, MIN_CONFIDENCE_FLOOR, MIN_CONFIDENCE_CEIL)

    if action in ("open_long", "open_short") and not has_pos:
        if decision["confidence"] < min_conf:
            record_step(store, "decision_skip_low_confidence",
                        {"action": action, "confidence": decision["confidence"], "min": min_conf}, "#999999")
            return
        open_position(store, decision, ticker, add_mode=False)

    elif action in ("open_long", "open_short") and has_pos and pos.get("direction") != decision["direction"]:
        close_position(store, ticker, "reverse_signal")
        Sleep(1000)
        if decision["confidence"] >= min_conf:
            open_position(store, decision, ticker, add_mode=False)

    elif action in ("add_long", "add_short") and has_pos and pos.get("direction") == decision["direction"]:
        if decision["confidence"] < min_conf:
            return
        if pos.get("add_count", 0) >= MAX_TOTAL_ADD_COUNT:
            record_step(store, "add_blocked_max_count", {"add_count": pos.get("add_count", 0)}, "#FFAA00")
            return
        open_position(store, decision, ticker, add_mode=True)

    elif action in ("close", "reduce") and has_pos:
        if action == "reduce":
            reduce_position(store, ticker, 0.5, "ai_reduce")
        else:
            close_position(store, ticker, "ai_decision")

    else:
        record_step(store, "decision_hold", {"action": action, "confidence": decision["confidence"]}, "#999999")


def open_position(store, decision, ticker, add_mode):
    market = store.get("market")
    if not market or not ticker:
        return False

    if TRADE_MODE != "trade":
        record_step(store, "trade_notify", {
            "action": decision["action"], "direction": decision["direction"],
            "confidence": decision["confidence"], "reason": decision["reason"],
        }, "#00CC66")
        return True

    acc = None
    try:
        acc = exchange.GetAccount()
    except Exception as e:
        Log("GetAccount失败: " + str(e), "#FF0000")
        return False
    if not acc or not ticker or ticker.get("Last", 0) <= 0:
        return False

    equity = get_equity(acc)
    base_pos_pct = playbook_param(store, "base_pos_pct", BASE_POS_PCT, BASE_POS_PCT_FLOOR, BASE_POS_PCT_CEIL)
    size_pct = base_pos_pct * clamp(decision["confidence"] / 100.0, 0.4, 1.0)
    qty = _N(equity * size_pct * LEVERAGE / ticker["Last"] / market["ctVal"], market["amountPrecision"])
    if qty <= 0 or qty < market["minQty"]:
        record_step(store, "trade_blocked", {"reason": "数量过小", "qty": qty}, "#FFAA00")
        return False

    pos = _G(POSITION_STATE_KEY) or {}
    cur_qty = pos.get("qty", 0) if add_mode else 0
    max_qty = _N(equity * MAX_SINGLE_POS_PCT * LEVERAGE / ticker["Last"] / market["ctVal"], market["amountPrecision"])
    if cur_qty + qty > max_qty:
        qty = _N(max(0, max_qty - cur_qty), market["amountPrecision"])
    if qty <= 0 or qty < market["minQty"]:
        record_step(store, "trade_blocked", {"reason": "已达单标的仓位上限"}, "#FFAA00")
        return False

    side = "buy" if decision["direction"] == "long" else "sell"
    oid = None
    try:
        oid = exchange.CreateOrder(market["contract"], side, -1, qty)
    except Exception as e:
        Log("下单异常: " + str(e), "#FF0000")
    if not oid:
        record_step(store, "trade_failed", {"side": side, "qty": qty}, "#FF0000")
        return False

    hard_stop_pct = playbook_param(store, "hard_stop_pct", HARD_STOP_PCT, HARD_STOP_PCT_MIN, HARD_STOP_PCT_MAX)
    now = int(time.time())

    if add_mode and pos:
        new_qty = pos.get("qty", 0) + qty
        avg_entry = ((pos.get("entry", ticker["Last"]) * pos.get("qty", 0) + ticker["Last"] * qty) / new_qty
                     if new_qty > 0 else ticker["Last"])
        pos["qty"] = new_qty
        pos["entry"] = avg_entry
        pos["add_count"] = pos.get("add_count", 0) + 1
    else:
        pos = {
            "direction": decision["direction"], "qty": qty, "entry": ticker["Last"],
            "opened_at": now, "add_count": 0, "peakPnlPct": 0, "trailActive": False,
            "decision_id": decision.get("id"),
        }

    pos["stop"] = (pos["entry"] * (1 - hard_stop_pct / 100.0) if pos["direction"] == "long"
                    else pos["entry"] * (1 + hard_stop_pct / 100.0))
    _G(POSITION_STATE_KEY, pos)

    if decision.get("id"):
        db_exec("UPDATE decisions SET executed=1 WHERE id=?", decision["id"])

    record_step(store, "trade_opened", {
        "direction": pos["direction"], "qty": qty, "price": ticker["Last"],
        "add_mode": add_mode, "hard_stop_pct": hard_stop_pct,
    }, "#00CC66")
    return True


def close_position(store, ticker, reason):
    pos = _G(POSITION_STATE_KEY)
    if not pos or not pos.get("qty"):
        return False
    market = store.get("market")
    if not market:
        return False

    if TRADE_MODE == "trade":
        side = "closebuy" if pos["direction"] == "long" else "closesell"
        oid = None
        try:
            oid = exchange.CreateOrder(market["contract"], side, -1, pos["qty"])
        except Exception as e:
            Log("平仓异常: " + str(e), "#FF0000")
        if not oid:
            record_step(store, "close_failed", {"reason": reason}, "#FF0000")
            return False

    exit_price = ticker["Last"] if ticker else pos.get("entry", 0)
    pnl_pct = (exit_price - pos["entry"]) / pos["entry"] * 100 * (1 if pos["direction"] == "long" else -1)
    now = int(time.time())

    db_exec('INSERT INTO trades(open_time, close_time, direction, entry_price, exit_price, qty, pnl_pct, '
            'close_reason, decision_id) values(?,?,?,?,?,?,?,?,?)',
            pos.get("opened_at", now), now, pos["direction"], pos["entry"], exit_price,
            pos["qty"], round(pnl_pct, 3), reason, pos.get("decision_id"))
    trade_id = db_last_id()

    record_step(store, "trade_closed", {
        "direction": pos["direction"], "pnl_pct": round(pnl_pct, 3), "reason": reason,
    }, "#00CC66" if pnl_pct > 0 else "#FF6666")

    _G(POSITION_STATE_KEY, None)

    if trade_id:
        reflect_on_closed_trade(store, trade_id)
        store["closed_trade_count_since_evolution"] = store.get("closed_trade_count_since_evolution", 0) + 1
        save_store(store)
        maybe_run_evolution(store)
    return True


def reduce_position(store, ticker, ratio, reason):
    pos = _G(POSITION_STATE_KEY)
    if not pos or not pos.get("qty"):
        return False
    market = store.get("market")
    if not market:
        return False
    cut = _N(max(0, pos["qty"] * clamp(ratio, 0.1, 1.0)), market["amountPrecision"])
    if cut <= 0 or cut < market.get("minQty", 0):
        return False
    if cut >= pos["qty"]:
        return close_position(store, ticker, reason)

    if TRADE_MODE == "trade":
        side = "closebuy" if pos["direction"] == "long" else "closesell"
        oid = None
        try:
            oid = exchange.CreateOrder(market["contract"], side, -1, cut)
        except Exception as e:
            Log("减仓异常: " + str(e), "#FF0000")
        if not oid:
            record_step(store, "reduce_failed", {"reason": reason}, "#FF0000")
            return False

    pos["qty"] = _N(pos["qty"] - cut, market["amountPrecision"])
    _G(POSITION_STATE_KEY, pos)
    record_step(store, "position_reduced", {"cut": cut, "left": pos["qty"], "reason": reason}, "#FFAA00")
    return True


def monitor_position(store):
    pos = _G(POSITION_STATE_KEY)
    if not pos or not pos.get("qty"):
        return
    market = store.get("market")
    if not market:
        return
    ticker = None
    try:
        ticker = exchange.GetTicker(market["contract"])
    except Exception as e:
        Log("GetTicker失败: " + str(e), "#FFAA00")
        return
    if not ticker or ticker.get("Last", 0) <= 0:
        return

    direction = pos["direction"]
    pnl_pct = (ticker["Last"] - pos["entry"]) / pos["entry"] * 100 * (1 if direction == "long" else -1)
    if pnl_pct > pos.get("peakPnlPct", 0):
        pos["peakPnlPct"] = pnl_pct

    trail_activate = playbook_param(store, "trail_activate_pct", TRAIL_ACTIVATE_PCT,
                                     TRAIL_ACTIVATE_PCT_MIN, TRAIL_ACTIVATE_PCT_MAX)
    trail_giveback = playbook_param(store, "trail_giveback_pct", TRAIL_GIVEBACK_PCT,
                                     TRAIL_GIVEBACK_PCT_MIN, TRAIL_GIVEBACK_PCT_MAX)

    if not pos.get("trailActive") and pos.get("peakPnlPct", 0) >= trail_activate:
        pos["trailActive"] = True

    hard_stop = ticker["Last"] <= pos["stop"] if direction == "long" else ticker["Last"] >= pos["stop"]

    trail_hit = False
    if pos.get("trailActive"):
        giveback = pos.get("peakPnlPct", 0) - pnl_pct
        allowed = max(1.0, pos.get("peakPnlPct", 0) * trail_giveback / 100.0)
        trail_hit = giveback >= allowed

    if hard_stop or trail_hit:
        close_position(store, ticker, "hard_stop" if hard_stop else "trailing_stop")
    else:
        _G(POSITION_STATE_KEY, pos)


def get_equity(acc):
    return acc.get("Equity") or acc.get("Balance") or acc.get("Stocks") or 0


# =========================
# 复盘层
# =========================

def reflect_on_closed_trade(store, trade_id):
    if not LLM_API_KEY:
        return
    rows = db_query("select * from trades where id=?", trade_id)
    if not rows:
        return
    trade = rows[0]
    entry_decision = None
    if trade.get("decision_id"):
        drows = db_query("select * from decisions where id=?", trade["decision_id"])
        entry_decision = drows[0] if drows else None

    schema = {
        "outcome": "win|loss|breakeven",
        "mistake_type": "none|chased_move|ignored_news_conflict|stop_too_tight|stop_too_wide|"
                         "overleveraged|held_too_long|exited_too_early|other",
        "lesson": "",
        "tags": [""],
    }

    prompt = "\n".join([
        "对以下已平仓交易进行复盘：",
        "交易记录:", json.dumps(trade, ensure_ascii=False),
        "入场时的决策依据:", json.dumps(entry_decision, ensure_ascii=False) if entry_decision else "无记录",
        "输出schema:", json.dumps(schema, ensure_ascii=False),
        "要求：",
        "1. outcome根据pnl_pct判断，明显为正=win，明显为负=loss，接近0=breakeven。",
        "2. lesson必须是具体、可执行的中文经验总结，不要空泛(例如'RSI超过75追多容易被套，应等回调')。",
        "3. mistake_type只在确实存在对应问题时才选择非none选项。",
        "4. 字段名必须和schema完全一致。",
    ])

    data = call_llm_json("你是交易复盘教练，只输出JSON，所有文本字段必须使用中文。", prompt, 900)
    if not data:
        return

    now = int(time.time())
    db_exec('INSERT INTO reflections(trade_id, time, outcome, mistake_type, lesson, tags, raw_json) '
            'values(?,?,?,?,?,?,?)',
            trade_id, now, data.get("outcome", ""), data.get("mistake_type", "none"),
            data.get("lesson", ""), json.dumps(as_list(data.get("tags")), ensure_ascii=False),
            json.dumps(data, ensure_ascii=False))
    db_exec('UPDATE trades SET reflected=1 WHERE id=?', trade_id)

    record_step(store, "reflection_saved", {
        "trade_id": trade_id, "outcome": data.get("outcome"),
        "lesson": short_text(data.get("lesson", ""), 80),
    }, "#00AAFF")


# =========================
# 进化层 (核心：自进化闭环)
# =========================

def maybe_run_evolution(store):
    if store.get("evolution_paused"):
        return
    total_reflected = db_query("select count(*) as c from reflections")
    total_reflected = total_reflected[0]["c"] if total_reflected else 0
    if total_reflected < EVOLUTION_MIN_TRADES:
        return

    now = int(time.time())
    closed_since = store.get("closed_trade_count_since_evolution", 0)
    trade_due = closed_since >= EVOLUTION_TRADE_INTERVAL
    time_due = now - store.get("last_evolution_check", 0) > EVOLUTION_TIME_INTERVAL_S

    if trade_due or time_due:
        run_evolution(store)


def run_evolution(store, forced=False):
    rows = db_query(
        "select r.*, t.direction as t_direction, t.pnl_pct as t_pnl_pct, t.close_reason as t_close_reason "
        "from reflections r left join trades t on r.trade_id = t.id order by r.time desc limit ?",
        REFLECTION_LOOKBACK)
    if not rows:
        if not forced:
            return
        record_step(store, "evolution_skipped", {"reason": "暂无复盘记录"}, "#FFAA00")
        return

    if not LLM_API_KEY:
        record_step(store, "evolution_skipped", {"reason": "LLM_API_KEY未配置"}, "#FFAA00")
        return

    stats = compute_trade_stats()
    playbook = store.get("current_playbook", {})

    schema = {
        "summary": "",
        "do_rules": [""],
        "avoid_rules": [""],
        "param_suggestions": {
            "min_confidence": 0,
            "base_pos_pct": 0.0,
            "hard_stop_pct": 0.0,
            "trail_activate_pct": 0.0,
            "trail_giveback_pct": 0.0,
        },
        "confidence_in_update": 0,
        "reasoning": "",
    }

    prompt = "\n".join([
        "你是交易系统的自我进化复盘员，任务是基于历史交易复盘更新交易playbook(经验库)。",
        "当前playbook:", json.dumps(playbook, ensure_ascii=False),
        "整体统计:", json.dumps(stats, ensure_ascii=False),
        "最近复盘记录:", json.dumps(rows[:REFLECTION_LOOKBACK], ensure_ascii=False),
        "输出schema:", json.dumps(schema, ensure_ascii=False),
        "要求：",
        "1. do_rules/avoid_rules必须是具体、可执行的中文规则，例如'RSI>70且新闻利空时不要开多'。",
        "2. param_suggestions只在有充分证据支持时才调整，避免频繁大幅震荡；证据不足时原样返回当前值。",
        "3. summary是对当前策略状态的整体中文总结，控制在150字以内。",
        "4. 字段名必须和schema完全一致，不要新增或改名。",
    ])

    data = call_llm_json("你是量化交易复盘与策略进化专家，只输出JSON，所有文本字段必须使用中文。", prompt, 1600)
    if not data:
        record_step(store, "evolution_failed", {"reason": "LLM返回空"}, "#FFAA00")
        return

    old_params = playbook.get("params", default_playbook_params())
    suggested = data.get("param_suggestions", {}) or {}
    new_params = dict(old_params)
    bounds = {
        "min_confidence": (MIN_CONFIDENCE_FLOOR, MIN_CONFIDENCE_CEIL),
        "base_pos_pct": (BASE_POS_PCT_FLOOR, BASE_POS_PCT_CEIL),
        "hard_stop_pct": (HARD_STOP_PCT_MIN, HARD_STOP_PCT_MAX),
        "trail_activate_pct": (TRAIL_ACTIVATE_PCT_MIN, TRAIL_ACTIVATE_PCT_MAX),
        "trail_giveback_pct": (TRAIL_GIVEBACK_PCT_MIN, TRAIL_GIVEBACK_PCT_MAX),
    }
    for k, (lo, hi) in bounds.items():
        if k in suggested:
            try:
                new_params[k] = round(clamp(float(suggested[k]), lo, hi), 4)
            except Exception:
                pass

    new_version = playbook.get("version", 0) + 1
    now = int(time.time())
    do_rules = as_list(data.get("do_rules"))[:12]
    avoid_rules = as_list(data.get("avoid_rules"))[:12]
    summary = data.get("summary", "") or ""

    db_exec('INSERT INTO playbook(version, created_at, summary, do_rules, avoid_rules, params_json, based_on_trades) '
            'values(?,?,?,?,?,?,?)',
            new_version, now, summary, json.dumps(do_rules, ensure_ascii=False),
            json.dumps(avoid_rules, ensure_ascii=False), json.dumps(new_params, ensure_ascii=False), len(rows))

    store["current_playbook"] = {
        "version": new_version, "summary": summary, "do_rules": do_rules,
        "avoid_rules": avoid_rules, "params": new_params, "based_on_trades": len(rows),
    }
    store["closed_trade_count_since_evolution"] = 0
    store["last_evolution_check"] = now
    save_store(store)

    record_step(store, "evolution_done", {
        "version": new_version, "summary": summary,
        "old_params": old_params, "new_params": new_params, "based_on_trades": len(rows),
    }, "#00CC66")


def compute_trade_stats():
    rows = db_query("select pnl_pct from trades where close_time is not null order by close_time desc limit 100")
    pnls = [r["pnl_pct"] for r in rows if r.get("pnl_pct") is not None]
    if not pnls:
        return {"total": 0, "win_rate": 0, "avg_pnl_pct": 0}
    wins = [p for p in pnls if p > 0]
    losses = [p for p in pnls if p <= 0]
    return {
        "total": len(pnls),
        "win_rate": round(len(wins) / len(pnls), 3),
        "avg_pnl_pct": round(mean(pnls), 3),
        "avg_win_pct": round(mean(wins), 3) if wins else 0,
        "avg_loss_pct": round(mean(losses), 3) if losses else 0,
    }


# =========================
# Dashboard
# =========================

def show_dashboard(store):
    market = store.get("market") or {}
    playbook = store.get("current_playbook", {})
    pos = _G(POSITION_STATE_KEY) or {}
    stats = compute_trade_stats()

    header = "{} 单品种AI自进化交易系统 | {} | mode:{} | playbook v{}\n".format(
        INSTRUMENT_DISPLAY_NAME, time.strftime("%Y-%m-%d %H:%M:%S"), TRADE_MODE, playbook.get("version", 0))

    overview_rows = [
        ["合约", market.get("contract", "-")],
        ["总交易数", stats.get("total", 0)],
        ["胜率", "{}%".format(round(stats.get("win_rate", 0) * 100, 1))],
        ["平均盈亏%", stats.get("avg_pnl_pct", 0)],
        ["当前持仓", "{}/{}".format(pos.get("direction", "-"), fmt(pos.get("qty", 0))) if pos else "无"],
        ["累计加仓次数", pos.get("add_count", 0) if pos else 0],
        ["距下次进化", "{}/{}笔".format(store.get("closed_trade_count_since_evolution", 0), EVOLUTION_TRADE_INTERVAL)],
        ["进化状态", "已暂停" if store.get("evolution_paused") else "运行中"],
    ]
    overview = make_table("系统概览", ["项目", "数值"], overview_rows)

    snap = store.get("indicator_snapshot") or {}
    ind_rows = []
    if snap:
        t, m = snap.get("trend", {}), snap.get("momentum", {})
        v, vo, pa = snap.get("volatility", {}), snap.get("volume", {}), snap.get("price_action", {})
        ind_rows = build_indicator_rows(snap, t, m, v, vo, pa)
    if not ind_rows:
        ind_rows = [["-", "-", "-", "-", "-", "-", "-"]]
    indicator_table = make_table(
        "📊 多指标快照",
        ["维度", "指标①", "数值①", "指标②", "数值②", "指标③", "数值③"],
        ind_rows)

    signal_rows = []
    for s in store.get("signals", [])[:8]:
        signal_rows.append([
            time.strftime("%H:%M:%S", time.localtime(s.get("time", 0))), s.get("action"), s.get("confidence"),
            s.get("technical_view"), s.get("news_view"), short_text(s.get("reason", ""), 40),
        ])
    if not signal_rows:
        signal_rows = [["-", "-", "-", "-", "-", "暂无"]]
    signals_table = make_table("最近决策", ["时间", "动作", "置信度", "技术面", "消息面", "理由"], signal_rows)

    trade_rows = []
    for t in db_query("select * from trades order by close_time desc limit 8"):
        trade_rows.append([
            time.strftime("%H:%M:%S", time.localtime(t.get("close_time") or 0)), t.get("direction"),
            fmt(t.get("entry_price")), fmt(t.get("exit_price")), fmt(t.get("pnl_pct")), t.get("close_reason"),
        ])
    if not trade_rows:
        trade_rows = [["-", "-", "-", "-", "-", "暂无"]]
    trades_table = make_table("最近交易", ["平仓时间", "方向", "入场", "出场", "盈亏%", "原因"], trade_rows)

    ref_rows = []
    for r in db_query("select * from reflections order by time desc limit 6"):
        ref_rows.append([
            time.strftime("%H:%M:%S", time.localtime(r.get("time") or 0)), r.get("outcome"),
            r.get("mistake_type"), short_text(r.get("lesson", ""), 60),
        ])
    if not ref_rows:
        ref_rows = [["-", "-", "-", "暂无"]]
    reflections_table = make_table("最近复盘", ["时间", "结果", "问题类型", "经验"], ref_rows)

    playbook_rows = [
        ["版本", playbook.get("version", 0)],
        ["摘要", short_text(playbook.get("summary", ""), 80)],
        ["Do规则", " | ".join([short_text(x, 30) for x in playbook.get("do_rules", [])[:5]]) or "-"],
        ["Avoid规则", " | ".join([short_text(x, 30) for x in playbook.get("avoid_rules", [])[:5]]) or "-"],
        ["当前参数", json.dumps(playbook.get("params", {}), ensure_ascii=False)],
    ]
    playbook_table = make_table("当前Playbook(经验库)", ["项目", "内容"], playbook_rows)

    step_rows = []
    for row in store.get("step_logs", [])[:8]:
        step_rows.append([
            time.strftime("%H:%M:%S", time.localtime(row.get("time", 0))), row.get("step", ""),
            short_text(json.dumps(row.get("data", {}), ensure_ascii=False), 90),
        ])
    if not step_rows:
        step_rows = [["-", "-", "暂无"]]
    steps_table = make_table("最近步骤", ["时间", "步骤", "内容"], step_rows)

    LogStatus(
        header +
        status_table(overview) +
        status_table(indicator_table) +
        status_table(signals_table) +
        status_table(trades_table) +
        status_table(reflections_table) +
        status_table(playbook_table) +
        status_table(steps_table)
    )


def build_indicator_rows(snap, t, m, v, vo, pa):
    price = snap.get("price")
    ema_f, ema_s = t.get("ema_fast"), t.get("ema_slow")
    bias = t.get("ema_bias_pct", 0)
    # 趋势方向标记：快线在慢线上方且价格在快线上方 -> 多头排列
    trend_tag = "多头↑" if (num(ema_f) > num(ema_s) and num(price) > num(ema_f)) else \
                ("空头↓" if (num(ema_f) < num(ema_s) and num(price) < num(ema_f)) else "震荡→")
    hist = t.get("macd_hist")
    macd_tag = "红↑" if num(hist) > 0 else "绿↓"
    rsi = m.get("rsi14")
    rsi_tag = rsi_state(rsi)
    k, d = m.get("kdj_k"), m.get("kdj_d")
    kdj_tag = "金叉↑" if num(k) > num(d) else "死叉↓"
    bp = v.get("band_position")
    bp_tag = band_state(bp)
    chg1 = pa.get("pct_change_1bar")
    chg6 = pa.get("pct_change_6bar")

    return [
        ["趋势",
         "价格", fmt_price(price),
         "EMA20", fmt_price(ema_f),
         "EMA60", fmt_price(ema_s)],
        ["趋势",
         "均线状态", trend_tag,
         "快慢乖离", fmt_pct(bias),
         "MACD柱", "{} {}".format(fmt_price(hist), macd_tag)],
        ["动量",
         "RSI14", "{} {}".format(fmt_num(rsi), rsi_tag),
         "KDJ_K", fmt_num(k),
         "KDJ状态", "{} {}".format(fmt_num(d), kdj_tag)],
        ["通道",
         "BOLL上", fmt_price(v.get("boll_upper")),
         "BOLL下", fmt_price(v.get("boll_lower")),
         "带宽位置", "{} {}".format(fmt_num(bp), bp_tag)],
        ["波动量能",
         "ATR%", fmt_pct(v.get("atr_pct")),
         "量比", fmt_num(vo.get("vol_ratio_to_ma20")),
         "近1/6根%", "{} / {}".format(fmt_pct(chg1), fmt_pct(chg6))],
    ]


def rsi_state(rsi):
    r = num(rsi)
    if r >= 70:
        return "超买"
    if r <= 30:
        return "超卖"
    if r >= 55:
        return "偏强"
    if r <= 45:
        return "偏弱"
    return "中性"


def band_state(bp):
    b = num(bp)
    if b >= 0.8:
        return "近上轨"
    if b <= 0.2:
        return "近下轨"
    if b >= 0.55:
        return "中上"
    if b <= 0.45:
        return "中下"
    return "居中"


def num(x):
    try:
        return float(x)
    except Exception:
        return 0.0


def fmt_price(x):
    v = num(x)
    if abs(v) >= 1000:
        return "{:,.1f}".format(v)
    if abs(v) >= 1:
        return "{:.3f}".format(v)
    return "{:.5f}".format(v)


def fmt_num(x):
    if x is None:
        return "-"
    return "{:.2f}".format(num(x))


def fmt_pct(x):
    if x is None:
        return "-"
    return "{:+.2f}%".format(num(x))


def make_table(title, cols, rows):
    return {"type": "table", "title": title, "cols": cols, "rows": rows}


def status_table(table):
    return "\n`" + json.dumps(table, ensure_ascii=False) + "`\n"


# =========================
# 工具函数
# =========================

def call_llm_json(system_prompt, user_prompt, max_tokens):
    body = json.dumps({
        "model": LLM_MODEL,
        "temperature": 0.1,
        "max_tokens": max_tokens,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
    })
    try:
        raw = http_request(LLM_BASE_URL, "POST", {
            "Content-Type": "application/json",
            "Authorization": "Bearer " + LLM_API_KEY,
        }, body, 40)
        data = json.loads(raw)
        if data.get("error"):
            Log("LLM错误: " + json.dumps(data["error"], ensure_ascii=False), "#FF0000")
            return None
        return json.loads(clean_json_text(data["choices"][0]["message"]["content"]))
    except Exception as e:
        Log("LLM解析失败: " + str(e), "#FFAA00")
        return None


def http_request(url, method, headers, body, timeout):
    data = body.encode("utf-8") if body else None
    req = urllib.request.Request(url, data=data, headers=headers or {}, method=method)
    resp = urllib.request.urlopen(req, timeout=timeout)
    try:
        return resp.read().decode("utf-8", errors="ignore")
    finally:
        resp.close()


def clean_json_text(text):
    text = re.sub(r"<think>[\s\S]*?</think>", "", text or "").strip()
    text = re.sub(r"```json\s*|```\s*", "", text).strip()
    first, last = text.find("{"), text.rfind("}")
    return text[first:last + 1] if first >= 0 and last > first else text


def normalize_confidence(value):
    try:
        v = float(value or 0)
    except Exception:
        return 0
    if 0 < v <= 1:
        v = v * 100
    return int(round(max(0, min(100, v))))


def safe_float(value, default=0.0):
    try:
        return float(value)
    except Exception:
        return default


def safe_json_loads(text, default):
    try:
        return json.loads(text) if text else default
    except Exception:
        return default


def as_list(x):
    if x is None:
        return []
    if isinstance(x, list):
        return x
    return [x]


def mean(arr):
    return sum(arr) / len(arr) if arr else 0.0


def std(arr):
    if len(arr) < 2:
        return 0.0
    m = mean(arr)
    return math.sqrt(sum((x - m) ** 2 for x in arr) / (len(arr) - 1))


def clamp(x, lo, hi):
    return max(lo, min(hi, x))


def short_text(text, n):
    text = str(text or "").replace("\n", " ").replace("\r", " ")
    return text[:n] + ("..." if len(text) > n else "")


def fmt(x):
    try:
        return round(float(x), 4)
    except Exception:
        return x