TabFM 原始K线预测交易策略 — 描述手册

# -*- coding: utf-8 -*-
"""
FMZ 发明者量化：TabFM 原始K线预测交易单文件示范

核心目标：
1. 这是发明者 FMZ Python 策略代码，可以直接复制到 FMZ 策略编辑器。
2. 启动时检查 Python 包，缺少则降级为 mock（不自动安装）。
3. 直接用 exchange.GetRecords() 获取 K 线。
4. 不计算技术指标，直接把最近一段 OHLCV K线展开为表格特征。
5. 用 TabFM 预测未来方向 up / down / flat 的概率。
6. 按概率阈值执行交易。

重要说明：
- 默认 AUTO_TRADE=False，只打印信号，不真实下单。
- 如果 FMZ 托管者环境未安装 tabfm，则自动降级为 mock 预测。
- 这是科普示范，不是可直接实盘的盈利策略。

建议 FMZ 参数：
SYMBOL: 字符串，例如 BTC_USDT.swap，可留空使用当前交易对
PERIOD_NAME: 字符串，例如 PERIOD_M15
BAR_LIMIT: 数字，默认 260
TRAIN_ROWS: 数字，默认 100，作为 TabFM 上下文的历史样本行数
RETURN_THRESHOLD: 数字，默认 0.0003，下一根K线涨跌分类阈值
MIN_CONFIDENCE: 数字，默认 0.60
USE_REAL_TABFM: 布尔，默认 true
AUTO_TRADE: 布尔，默认 false
RISK_PCT: 数字，默认 0.01
MAX_LEVERAGE: 数字，默认 2
HARD_STOP_ATR: 数字，默认 2.0
TRAIL_START_ATR: 数字，默认 2.0，盈利超过多少 ATR 后启动移动止损
TRAIL_GIVEBACK_ATR: 数字，默认 1.0，启动后允许回撤多少 ATR
REVERSE_ON_SIGNAL: 布尔，默认 true，反向信号出现时先平仓再开仓
CHECK_INTERVAL_S: 数字，默认 10
COOLDOWN_BARS: 数字，默认 3

运行频率：
- 策略每次启动后立即执行一次 TabFM 预测。
- 此后只在每个自然小时整点执行一次预测。
- 持仓止损与移动止盈仍按 CHECK_INTERVAL_S 持续监控。
"""

import os

# 必须在导入 JAX / TabFM 之前设置，避免 JAX 预占内存。
os.environ.setdefault("JAX_PLATFORMS", "cpu")
os.environ.setdefault("XLA_PYTHON_CLIENT_PREALLOCATE", "false")
os.environ.setdefault("XLA_PYTHON_CLIENT_ALLOCATOR", "platform")

import importlib
import json
import math
import time


STORE_KEY = "fmz_tabfm_onefile_strategy_v1"
STRATEGY_NAME = "fmz_tabfm_raw_kline_predictor"
MODEL_CACHE = {
    "model": None,
    "classifier": None,
    "available": None,
    "backend": "jax",
}

RAW_WINDOW_DEFAULT = 15
TRAIN_ROWS_DEFAULT = 100


def make_feature_columns(window):
    columns = []
    for lag in range(1, window + 1):
        prefix = "lag_{}".format(lag)
        columns.extend([
            prefix + "_open",
            prefix + "_high",
            prefix + "_low",
            prefix + "_close",
            prefix + "_volume",
        ])
    return columns


FEATURE_COLUMNS = make_feature_columns(RAW_WINDOW_DEFAULT)


# ----------------------------- FMZ 入口 -----------------------------


def main():
    cfg = load_config()
    store = load_store()
    Log("TabFM K线窗口预测策略启动 | symbol:{} | real_tabfm:{} | auto_trade:{}".format(
        cfg["symbol"] or "当前交易对", cfg["use_real_tabfm"], cfg["auto_trade"]
    ), "#00AAFF")

    try:
        check_runtime(cfg)
    except Exception as e:
        Log("环境检查异常（不影响运行）: " + repr(e), "#FF0000")

    # 无论持久化状态如何，每次策略进程启动都立即预测一次。
    try:
        run_prediction_cycle(store, cfg, trigger="startup")
        store["last_prediction_hour"] = current_hour_key()
        save_store(store)
    except Exception as e:
        Log("启动预测异常: " + repr(e), "#FF0000")

    Log("进入主循环：每小时整点预测一次，持仓持续监控", "#00AAFF")
    while True:
        try:
            handle_command(store)
            tick(store, cfg)
            show_dashboard(store, cfg)
            save_store(store)
        except Exception as e:
            Log("主循环异常: " + repr(e), "#FF0000")
        Sleep(int(cfg["check_interval_s"]) * 1000)


# ----------------------------- 配置和状态 -----------------------------


def g(name, default):
    return globals().get(name, default)


def load_config():
    return {
        "symbol": g("SYMBOL", ""),
        "period_name": g("PERIOD_NAME", "PERIOD_M15"),
        "bar_limit": int(g("BAR_LIMIT", 260)),
        "raw_window": int(g("RAW_WINDOW", RAW_WINDOW_DEFAULT)),
        "train_rows": int(g("TRAIN_ROWS", TRAIN_ROWS_DEFAULT)),
        "return_threshold": float(g("RETURN_THRESHOLD", 0.0003)),
        "min_confidence": float(g("MIN_CONFIDENCE", 0.60)),
        "use_real_tabfm": bool(g("USE_REAL_TABFM", True)),
        "auto_trade": bool(g("AUTO_TRADE", False)),
        "risk_pct": float(g("RISK_PCT", 0.01)),
        "max_leverage": float(g("MAX_LEVERAGE", 2)),
        "hard_stop_atr": float(g("HARD_STOP_ATR", 2.0)),
        "trail_start_atr": float(g("TRAIL_START_ATR", 2.0)),
        "trail_giveback_atr": float(g("TRAIL_GIVEBACK_ATR", 1.0)),
        "reverse_on_signal": bool(g("REVERSE_ON_SIGNAL", True)),
        "check_interval_s": int(g("CHECK_INTERVAL_S", 10)),
        "cooldown_bars": int(g("COOLDOWN_BARS", 3)),
    }


def new_store():
    return {
        "last_bar_time": 0,
        "last_prediction_hour": "",
        "last_prediction_at": 0,
        "last_prediction_trigger": "",
        "last_signal_bar": 0,
        "last_prediction": None,
        "last_action": "",
        "position_state": {},
        "signals": [],
        "logs": [],
        "runtime": {},
    }


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


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


def handle_command(store):
    cmd = GetCommand()
    if not cmd:
        return
    if cmd == "clear:store":
        store.clear()
        store.update(new_store())
        save_store(store)
        Log("已清空本地状态", "#FFAA00")
    elif cmd == "close:position":
        pos = get_position()
        if pos:
            close_position(pos, "manual_command")


# ----------------------------- 包检查和安装 -----------------------------


def check_runtime(cfg):
    status = {"mode": "mock", "packages": {}}
    for import_name, pip_name in [
        ("numpy", "numpy"),
        ("pandas", "pandas"),
    ]:
        status["packages"][import_name] = ensure_package(import_name)

    if cfg["use_real_tabfm"]:
        status["packages"]["tabfm"] = ensure_tabfm()
        if status["packages"]["tabfm"]:
            status["mode"] = "real_tabfm"

    store = load_store()
    store["runtime"] = status
    save_store(store)
    Log("运行环境检查完成: " + json.dumps(status, ensure_ascii=False), "#00AAFF")


def ensure_package(import_name):
    try:
        importlib.import_module(import_name)
        return True
    except Exception as exc:
        Log("缺少包 {}，请在托管者环境预先安装: {}".format(import_name, exc), "#FFAA00")
        return False


def ensure_tabfm():
    try:
        import tabfm
        from tabfm import TabFMClassifier
        from tabfm import tabfm_v1_0_0_jax

        Log(
            "检测到 TabFM JAX 后端 | tabfm路径: {}".format(
                getattr(tabfm, "__file__", "unknown")
            ),
            "#00AAFF",
        )
        return True
    except Exception as exc:
        Log(
            "未检测到可用的 TabFM JAX 后端，降级 mock: {}".format(exc),
            "#FFAA00",
        )
        return False


# ----------------------------- 主循环 -----------------------------


def current_hour_key(now_ts=None):
    """返回本地时间的小时键，例如 2026-191-10。"""
    now_ts = time.time() if now_ts is None else now_ts
    tm = time.localtime(now_ts)
    return "{:04d}-{:03d}-{:02d}".format(tm.tm_year, tm.tm_yday, tm.tm_hour)


def next_hour_timestamp(now_ts=None):
    now_ts = time.time() if now_ts is None else now_ts
    tm = time.localtime(now_ts)
    current_hour_start = time.mktime((
        tm.tm_year, tm.tm_mon, tm.tm_mday, tm.tm_hour, 0, 0,
        tm.tm_wday, tm.tm_yday, tm.tm_isdst,
    ))
    return int(current_hour_start + 3600)


def tick(store, cfg):
    """
    主循环每 CHECK_INTERVAL_S 调用一次：
    1. 每个自然小时只触发一次 TabFM 预测；
    2. 每次循环都执行持仓监控。
    """
    hour_key = current_hour_key()

    if store.get("last_prediction_hour") != hour_key:
        # 第一次跨入新小时即执行，通常在整点后的 0～CHECK_INTERVAL_S 秒内。
        store["last_prediction_hour"] = hour_key
        save_store(store)
        run_prediction_cycle(store, cfg, trigger="hourly")

    bars = get_records(
        cfg["symbol"],
        period_value(cfg["period_name"]),
        cfg["bar_limit"],
    )

    if bars:
        monitor_position(bars, cfg, store)


def run_prediction_cycle(store, cfg, trigger):
    bars = get_records(
        cfg["symbol"],
        period_value(cfg["period_name"]),
        cfg["bar_limit"],
    )

    need = cfg["raw_window"] + cfg["train_rows"]
    if len(bars) < need:
        Log("K线数量不足: {} / {}".format(len(bars), need), "#999999")
        store["last_action"] = "prediction_skipped_insufficient_bars"
        store["last_prediction_at"] = int(time.time())
        store["last_prediction_trigger"] = trigger
        return False

    started_at = time.time()
    last_bar = bars[-1]
    store["last_bar_time"] = last_bar["Time"]

    Log(
        "开始执行 TabFM 预测 | trigger:{} | bar_time:{}".format(
            trigger, last_bar["Time"]
        ),
        "#FFAA00",
    )

    rows, labels, latest = build_dataset(
        bars,
        cfg["raw_window"],
        cfg["train_rows"],
        cfg["return_threshold"],
    )

    prediction = predict_direction(rows, labels, latest, cfg)
    prediction["trigger"] = trigger
    prediction["started_at"] = int(started_at)
    prediction["finished_at"] = int(time.time())
    prediction["elapsed_seconds"] = round(time.time() - started_at, 3)

    store["last_prediction"] = prediction
    store["last_prediction_at"] = prediction["finished_at"]
    store["last_prediction_trigger"] = trigger

    Log(
        "TabFM 预测完成 | label:{} | confidence:{:.1%} | mode:{} | 耗时:{:.1f}s".format(
            prediction["label"],
            prediction["confidence"],
            prediction["mode"],
            prediction["elapsed_seconds"],
        ),
        "#00CC66",
    )

    signal = build_signal(store, bars, prediction, cfg)
    if signal:
        store["signals"].insert(0, signal)
        store["signals"] = store["signals"][:50]
        execute_signal(signal, cfg, store)

    monitor_position(bars, cfg, store)
    return True


# ----------------------------- 预测逻辑 -----------------------------


def predict_direction(rows, labels, latest, cfg):
    use_real = cfg["use_real_tabfm"] and can_use_tabfm()
    if use_real:
        try:
            label, probs = tabfm_predict(rows, labels, latest)
            mode = "real_tabfm"
        except Exception as exc:
            Log("TabFM 预测失败，降级 mock: " + str(exc), "#FF0000")
            label, probs = mock_predict(latest)
            mode = "mock_fallback"
    else:
        label, probs = mock_predict(latest)
        mode = "mock"

    return {
        "ok": True,
        "mode": mode,
        "label": label,
        "confidence": float(probs.get(label, 0)),
        "probabilities": probs,
        "latest_features": latest,
        "context_rows": len(rows),
        "label_counts": count_labels(labels),
    }


def can_use_tabfm():
    if MODEL_CACHE["available"] is False:
        return False

    try:
        from tabfm import TabFMClassifier
        from tabfm import tabfm_v1_0_0_jax

        if TabFMClassifier is None or tabfm_v1_0_0_jax is None:
            raise ImportError("TabFM JAX 后端不可用")

        MODEL_CACHE["available"] = True
        return True

    except Exception as exc:
        MODEL_CACHE["available"] = False
        Log("TabFM JAX 后端不可用: " + str(exc), "#FFAA00")
        return False


def tabfm_predict(rows, labels, latest):
    import numpy as np
    import pandas as pd
    from tabfm import TabFMClassifier
    from tabfm import tabfm_v1_0_0_jax as tabfm_backend

    if MODEL_CACHE["model"] is None:
        Log("加载 TabFM JAX 预训练模型，首次运行可能较慢...", "#FFAA00")
        MODEL_CACHE["model"] = tabfm_backend.load(
            model_type="classification"
        )
        Log("TabFM JAX 预训练模型加载完成", "#00CC66")

    if MODEL_CACHE["classifier"] is None:
        MODEL_CACHE["classifier"] = TabFMClassifier(
            model=MODEL_CACHE["model"],
            random_state=42,
        )

    clf = MODEL_CACHE["classifier"]
    feature_columns = list(rows[0].keys())
    x_train = pd.DataFrame(rows, columns=feature_columns)
    y_train = np.asarray(labels)
    x_test = pd.DataFrame([latest], columns=feature_columns)

    # 每小时更新一次上下文；只调用 predict_proba，避免 predict 再做一次推理。
    clf.fit(x_train, y_train)
    probabilities = clf.predict_proba(x_test)[0]

    probs = {
        str(cls): float(prob)
        for cls, prob in zip(clf.classes_, probabilities)
    }

    for class_name in ("up", "flat", "down"):
        probs.setdefault(class_name, 0.0)

    total = sum(probs.values())
    if total > 0:
        probs = {key: value / total for key, value in probs.items()}

    label = max(probs, key=probs.get)
    return label, probs


def mock_predict(latest):
    # mock 只用于环境缺失时维持策略运行，不代表 TabFM 结果。
    recent_return = float(latest.get("lag_1_close", 0.0) - latest.get("lag_4_close", 0.0))
    candle_pressure = float(latest.get("lag_1_close", 0.0) - latest.get("lag_1_open", 0.0))
    score = max(-20.0, min(20.0, (recent_return + candle_pressure) * 80.0))
    up = 1.0 / (1.0 + math.exp(-score))
    down = 1.0 - up
    flat = max(0.15, 1.0 - abs(up - down))
    total = up + down + flat
    probs = {"up": up / total, "flat": flat / total, "down": down / total}
    label = max(probs, key=probs.get)
    return label, probs

def count_labels(labels):
    out = {}
    for label in labels:
        out[label] = out.get(label, 0) + 1
    return out


# ----------------------------- 特征工程 -----------------------------


def build_dataset(bars, window, train_rows, return_threshold):
    """
    用过去 window 根已完成K线预测紧接着的下一根K线。

    训练样本按“最新到最旧”排列：
    rows[0] 是最近一条已知样本，rows[-1] 是最旧样本。

    每一行内部也按“最近到最远”排列：
    lag_1 是目标K线前一根，lag_window 是更早的一根。
    """
    minimum_bars = window + train_rows
    if len(bars) < minimum_bars:
        raise ValueError(
            "K线数量不足，至少需要 {} 根，当前 {} 根".format(
                minimum_bars, len(bars)
            )
        )

    rows = []
    labels = []

    # bars[-1] 是最新一根已经完成的K线，可以作为最近训练样本的标签。
    latest_known_target_idx = len(bars) - 1

    # 从最近的已知样本开始，逐行向历史方向移动。
    for offset in range(train_rows):
        target_idx = latest_known_target_idx - offset

        rows.append(
            make_raw_kline_row(
                bars,
                target_idx=target_idx,
                window=window,
            )
        )
        labels.append(
            make_next_bar_label(
                bars,
                target_idx=target_idx,
                threshold=return_threshold,
            )
        )

    # target_idx=len(bars) 表示下一根尚未出现的K线。
    # 特征只读取 bars[-1] 到 bars[-window]，不存在未来数据泄漏。
    latest = make_raw_kline_row(
        bars,
        target_idx=len(bars),
        window=window,
    )

    return rows, labels, latest


def make_next_bar_label(bars, target_idx, threshold):
    """根据目标K线相对前一根收盘价的收益生成 up/down/flat 标签。"""
    previous_close = float(bars[target_idx - 1]["Close"])
    target_close = float(bars[target_idx]["Close"])

    if previous_close <= 0:
        return "flat"

    target_return = target_close / previous_close - 1.0

    if target_return > threshold:
        return "up"
    if target_return < -threshold:
        return "down"
    return "flat"


def make_raw_kline_row(bars, target_idx, window):
    """
    使用目标K线之前的 window 根已完成K线构造一行表格。

    lag_1：距离目标最近的一根K线；
    lag_15：人工时间窗口中最早的一根K线。

    这个窗口显式保留局部时序位置，但 TabFM 仍将其视为固定列的表格，
    它不是 TimesFM 那种直接建模序列的原生时序基础模型。
    """
    anchor_close = float(bars[target_idx - 1]["Close"])
    if anchor_close <= 0:
        anchor_close = 1.0

    volumes = [
        float(bars[target_idx - lag].get("Volume", 0.0))
        for lag in range(1, window + 1)
    ]
    mean_volume = sum(volumes) / len(volumes) if volumes else 1.0
    if mean_volume <= 0:
        mean_volume = 1.0

    row = {}
    for lag in range(1, window + 1):
        bar = bars[target_idx - lag]
        prefix = "lag_{}".format(lag)

        row[prefix + "_open"] = safe_div(float(bar["Open"]) - anchor_close, anchor_close)
        row[prefix + "_high"] = safe_div(float(bar["High"]) - anchor_close, anchor_close)
        row[prefix + "_low"] = safe_div(float(bar["Low"]) - anchor_close, anchor_close)
        row[prefix + "_close"] = safe_div(float(bar["Close"]) - anchor_close, anchor_close)
        row[prefix + "_volume"] = safe_div(float(bar.get("Volume", 0.0)), mean_volume)

    return row


# ----------------------------- 交易逻辑 -----------------------------


def build_signal(store, bars, prediction, cfg):
    label = prediction["label"]
    confidence = float(prediction["confidence"])
    close = float(bars[-1]["Close"])
    atr_v = atr(bars, 14)[-1]
    bar_time = bars[-1]["Time"]

    cooldown_ms = cfg["cooldown_bars"] * period_ms(period_value(cfg["period_name"]))
    if bar_time - store.get("last_signal_bar", 0) < cooldown_ms:
        store["last_action"] = "cooldown"
        return None

    if confidence < cfg["min_confidence"]:
        store["last_action"] = "hold_low_confidence"
        return None

    if label == "up":
        direction = "long"
    elif label == "down":
        direction = "short"
    else:
        store["last_action"] = "hold_flat"
        return None

    store["last_signal_bar"] = bar_time
    return {
        "time": int(time.time()),
        "bar_time": bar_time,
        "strategy": STRATEGY_NAME,
        "direction": direction,
        "price": close,
        "atr": atr_v,
        "label": label,
        "confidence": confidence,
        "probabilities": prediction["probabilities"],
        "reason": "预测 {}，置信度 {:.1%}".format(label, confidence),
    }


def execute_signal(signal, cfg, store):
    Log("交易信号: {} | conf:{:.1%} | price:{}".format(signal["direction"], signal["confidence"], signal["price"]), "#00CC66")
    pos = get_position()
    if not cfg["auto_trade"]:
        store["last_action"] = "notify_" + signal["direction"]
        return True

    if pos:
        pos_dir = position_direction(pos)
        if pos_dir == signal["direction"]:
            store["last_action"] = "skip_same_position"
            return False
        if not cfg.get("reverse_on_signal", True):
            store["last_action"] = "skip_reverse_disabled"
            return False
        close_position(pos, "reverse_signal")
        store["position_state"] = {}
        Sleep(1000)

    ok = open_position(signal, cfg)
    if ok:
        init_position_state(store, signal)
    store["last_action"] = "open_" + signal["direction"] if ok else "open_failed"
    return ok


def open_position(signal, cfg):
    acc = exchange.GetAccount()
    ticker = get_ticker(cfg["symbol"])
    if not acc or not ticker or float(ticker["Last"]) <= 0:
        return False
    last = float(ticker["Last"])
    equity = get_equity(acc)
    risk_cash = equity * cfg["risk_pct"]
    stop_distance = max(float(signal["atr"]) * cfg["hard_stop_atr"], last * 0.001)
    qty = risk_cash / stop_distance
    max_qty = equity * cfg["max_leverage"] / last
    qty = normalize_qty(max(0.0, min(qty, max_qty)))
    if qty <= 0:
        return False

    if signal["direction"] == "long":
        exchange.SetDirection("buy")
        order_id = exchange.Buy(-1, qty)
    else:
        exchange.SetDirection("sell")
        order_id = exchange.Sell(-1, qty)
    Log("下单: {} qty:{} order:{}".format(signal["direction"], qty, order_id), "#00AAFF")
    return bool(order_id)


def monitor_position(bars, cfg, store=None):
    pos = get_position()
    if not pos:
        if store is not None:
            store["position_state"] = {}
        return
    ticker = get_ticker(cfg["symbol"])
    if not ticker or float(ticker["Last"]) <= 0:
        return

    atr_v = atr(bars, 14)[-1]
    direction = position_direction(pos)
    entry = position_price(pos)
    last = float(ticker["Last"])
    state = ensure_position_state(store, pos, direction, entry, last) if store is not None else {}

    hard_stop_distance = atr_v * cfg["hard_stop_atr"]
    if direction == "long":
        state["best_price"] = max(float(state.get("best_price", entry)), last)
        hard_stop = entry - hard_stop_distance
        profit_atr = safe_div(state["best_price"] - entry, atr_v)
        trail_stop = None
        if profit_atr >= cfg["trail_start_atr"]:
            trail_stop = state["best_price"] - atr_v * cfg["trail_giveback_atr"]
            state["trail_active"] = True
        final_stop = max(hard_stop, trail_stop) if trail_stop is not None else hard_stop
        state["stop_price"] = final_stop
        if last <= final_stop:
            close_position(pos, "trailing_stop" if state.get("trail_active") else "hard_stop")
            if store is not None:
                store["last_action"] = "closed_by_stop"
                store["position_state"] = {}
            return
    else:
        state["best_price"] = min(float(state.get("best_price", entry)), last)
        hard_stop = entry + hard_stop_distance
        profit_atr = safe_div(entry - state["best_price"], atr_v)
        trail_stop = None
        if profit_atr >= cfg["trail_start_atr"]:
            trail_stop = state["best_price"] + atr_v * cfg["trail_giveback_atr"]
            state["trail_active"] = True
        final_stop = min(hard_stop, trail_stop) if trail_stop is not None else hard_stop
        state["stop_price"] = final_stop
        if last >= final_stop:
            close_position(pos, "trailing_stop" if state.get("trail_active") else "hard_stop")
            if store is not None:
                store["last_action"] = "closed_by_stop"
                store["position_state"] = {}
            return

    state["last_price"] = last
    state["atr"] = atr_v
    if store is not None:
        store["position_state"] = state


def init_position_state(store, signal):
    store["position_state"] = {
        "direction": signal["direction"],
        "entry_price": float(signal["price"]),
        "best_price": float(signal["price"]),
        "last_price": float(signal["price"]),
        "atr": float(signal["atr"]),
        "stop_price": 0.0,
        "trail_active": False,
        "opened_at": int(time.time()),
    }


def ensure_position_state(store, pos, direction, entry, last):
    if store is None:
        return {}
    state = store.get("position_state") or {}
    if state.get("direction") != direction or abs(float(state.get("entry_price", entry)) - entry) > max(entry * 0.001, 1e-8):
        state = {
            "direction": direction,
            "entry_price": entry,
            "best_price": last,
            "last_price": last,
            "atr": 0.0,
            "stop_price": 0.0,
            "trail_active": False,
            "opened_at": int(time.time()),
        }
    return state


def close_position(pos, reason):
    amount = abs(float(pos.get("Amount", pos.get("amount", 0))))
    if amount <= 0:
        return False
    direction = position_direction(pos)
    if direction == "long":
        exchange.SetDirection("closebuy")
        order_id = exchange.Sell(-1, amount)
    else:
        exchange.SetDirection("closesell")
        order_id = exchange.Buy(-1, amount)
    Log("平仓: {} amount:{} reason:{} order:{}".format(direction, amount, reason, order_id), "#FFAA00")
    return bool(order_id)


# ----------------------------- FMZ 兼容工具 -----------------------------


def get_records(symbol, period, limit):
    try:
        if symbol:
            records = exchange.GetRecords(symbol, period, limit)
        else:
            records = exchange.GetRecords(period)
    except Exception:
        records = exchange.GetRecords(period)
    if not records:
        return []
    return records[-limit:]


def get_ticker(symbol):
    try:
        if symbol:
            return exchange.GetTicker(symbol)
        return exchange.GetTicker()
    except Exception:
        return exchange.GetTicker()


def get_position():
    try:
        positions = exchange.GetPositions(SYMBOL) if SYMBOL else exchange.GetPositions()
    except Exception:
        return None
    if not positions:
        return None
    for pos in positions:
        amount = abs(float(pos.get("Amount", pos.get("amount", 0))))
        if amount > 0:
            return pos
    return None


def position_direction(pos):
    pos_type = pos.get("Type", pos.get("type"))
    if pos_type in (0, "buy", "long", "LONG"):
        return "long"
    if pos_type in (1, "sell", "short", "SHORT"):
        return "short"
    return "long" if float(pos.get("Amount", 0)) >= 0 else "short"


def position_price(pos):
    return float(pos.get("Price", pos.get("price", pos.get("AvgPrice", 0))))


def get_equity(acc):
    for key in ("Equity", "Balance", "NetAsset"):
        if key in acc and acc[key] is not None:
            return float(acc[key])
    return 0.0


def normalize_qty(qty):
    if qty <= 0:
        return 0.0
    return math.floor(qty * 10000) / 10000.0


def period_value(period_name):
    return globals().get(period_name, PERIOD_M15)


def period_ms(period):
    mapping = {
        PERIOD_M1: 60000,
        PERIOD_M5: 300000,
        PERIOD_M15: 900000,
        PERIOD_M30: 1800000,
        PERIOD_H1: 3600000,
        PERIOD_D1: 86400000,
    }
    return mapping.get(period, 900000)


# ----------------------------- 指标工具 -----------------------------


def ema(values, period):
    if not values:
        return []
    alpha = 2.0 / (period + 1.0)
    out = [float(values[0])]
    for value in values[1:]:
        out.append(alpha * float(value) + (1.0 - alpha) * out[-1])
    return out


def sma(values, period):
    out = []
    for idx in range(len(values)):
        start = max(0, idx - period + 1)
        window = values[start : idx + 1]
        out.append(sum(window) / len(window))
    return out


def rolling_std(values, period):
    out = []
    for idx in range(len(values)):
        start = max(0, idx - period + 1)
        window = values[start : idx + 1]
        mean = sum(window) / len(window)
        variance = sum((x - mean) ** 2 for x in window) / len(window)
        out.append(math.sqrt(variance))
    return out


def rsi(closes, period):
    gains = [0.0]
    losses = [0.0]
    for idx in range(1, len(closes)):
        change = closes[idx] - closes[idx - 1]
        gains.append(max(change, 0.0))
        losses.append(max(-change, 0.0))
    avg_gain = sma(gains, period)
    avg_loss = sma(losses, period)
    out = []
    for gain, loss in zip(avg_gain, avg_loss):
        if loss <= 1e-12:
            out.append(100.0 if gain > 0 else 50.0)
        else:
            out.append(100.0 - 100.0 / (1.0 + gain / loss))
    return out


def atr(bars, period):
    trs = []
    prev_close = None
    for bar in bars:
        high = float(bar["High"])
        low = float(bar["Low"])
        close = float(bar["Close"])
        if prev_close is None:
            tr = high - low
        else:
            tr = max(high - low, abs(high - prev_close), abs(low - prev_close))
        trs.append(tr)
        prev_close = close
    return sma(trs, period)


def pct_change(values, idx, lag):
    if idx - lag < 0 or values[idx - lag] == 0:
        return 0.0
    return values[idx] / values[idx - lag] - 1.0


def safe_div(a, b):
    if abs(b) <= 1e-12:
        return 0.0
    return a / b


# ----------------------------- 展示 -----------------------------


def format_time(ts):
    if not ts:
        return "-"
    return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))


def show_dashboard(store, cfg):
    prediction = store.get("last_prediction") or {}
    probs = prediction.get("probabilities", {})
    runtime = store.get("runtime", {})
    table = {
        "type": "table",
        "title": "TabFM 原始K线预测交易示范",
        "cols": ["项目", "值"],
        "rows": [
            ["symbol", cfg["symbol"] or "当前交易对"],
            ["raw_window", cfg.get("raw_window", RAW_WINDOW_DEFAULT)],
            ["train_rows", cfg.get("train_rows", TRAIN_ROWS_DEFAULT)],
            ["mode", prediction.get("mode", runtime.get("mode", "-"))],
            ["schedule", "启动立即执行；之后每小时整点"],
            ["last_trigger", store.get("last_prediction_trigger", "-")],
            ["last_prediction_at", format_time(store.get("last_prediction_at", 0))],
            ["next_prediction_at", format_time(next_hour_timestamp())],
            ["elapsed_seconds", prediction.get("elapsed_seconds", "-")],
            ["auto_trade", cfg["auto_trade"]],
            ["label", prediction.get("label", "-")],
            ["confidence", "{:.1%}".format(float(prediction.get("confidence", 0)))],
            ["up", "{:.1%}".format(float(probs.get("up", 0)))],
            ["flat", "{:.1%}".format(float(probs.get("flat", 0)))],
            ["down", "{:.1%}".format(float(probs.get("down", 0)))],
            ["context_rows", prediction.get("context_rows", "-")],
            ["label_counts", json.dumps(prediction.get("label_counts", {}), ensure_ascii=False)],
            ["last_action", store.get("last_action", "")],
            ["trail_active", (store.get("position_state") or {}).get("trail_active", False)],
            ["best_price", (store.get("position_state") or {}).get("best_price", "-")],
            ["stop_price", (store.get("position_state") or {}).get("stop_price", "-")],
        ],
    }
    LogStatus("`" + json.dumps(table, ensure_ascii=False) + "`")