实时新闻导火线系统

# -*- coding: utf-8 -*-
"""
FUSE — Manual News Trading Panel (FMZ Python)
直接在 FMZ 平台运行，无需外部服务。
"""

import json
import time
import urllib.error
import urllib.request
import calendar

# ─── 常量映射 ─────────────────────────────────────────────────────────────────

PERIOD_MAP = {
    "1m":  PERIOD_M1,  "3m":  PERIOD_M3,  "5m":  PERIOD_M5,
    "15m": PERIOD_M15, "30m": PERIOD_M30,
    "1h":  PERIOD_H1,  "4h":  PERIOD_H4,  "1d":  PERIOD_D1,
}

PERIOD_MS = {
    "1m":  60000,    "3m":  180000,   "5m":  300000,
    "15m": 900000,   "30m": 1800000,
    "1h":  3600000,  "4h":  14400000, "1d":  86400000,
}

# ─── 状态 ─────────────────────────────────────────────────────────────────────

_chart          = None
_last_symbol    = ""
_cur_amount     = ORDER_AMOUNT
_cached_news    = []
_last_news_at   = 0
_mcp_session_id = ""
_mcp_req_id     = 0
_mcp_ready      = False
_init_equity    = None


# ─── MCP Client ───────────────────────────────────────────────────────────────

def _mcp_headers():
    h = {
        "Content-Type": "application/json",
        "Accept":        "application/json, text/event-stream",
        "Authorization": "Bearer " + JIN10_MCP_TOKEN,
        "User-Agent":    "FUSE-FMZ/1.0",
    }
    if _mcp_session_id:
        h["Mcp-Session-Id"] = _mcp_session_id
    return h


def _mcp_parse(text):
    last = None
    for line in text.splitlines():
        line = line.strip()
        if not line.startswith("data:"):
            continue
        chunk = line[5:].strip()
        if not chunk or chunk == "[DONE]":
            continue
        try:
            parsed = json.loads(chunk)
            if isinstance(parsed, dict):
                last = parsed
        except json.JSONDecodeError:
            pass
    if last is None:
        try:
            last = json.loads(text)
        except json.JSONDecodeError:
            return {"raw": text[:2000]}
    if "error" in last:
        raise RuntimeError("MCP error: " + str(last["error"]))
    return last.get("result", last)


def _mcp_post(payload, is_notification=False):
    global _mcp_session_id, _mcp_req_id
    if not is_notification:
        _mcp_req_id += 1
        payload["id"] = _mcp_req_id
    body = json.dumps(payload, ensure_ascii=False).encode()
    req  = urllib.request.Request(
        JIN10_MCP_URL, data=body, headers=_mcp_headers(), method="POST"
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            sid = resp.headers.get("Mcp-Session-Id")
            if sid:
                _mcp_session_id = sid
            if resp.status == 202:
                return {}
            text = resp.read().decode("utf-8", errors="replace")
    except urllib.error.HTTPError as e:
        raise RuntimeError("HTTP %d: %s" % (e.code, e.read().decode()[:400]))
    except urllib.error.URLError as e:
        raise RuntimeError("Network: " + str(e))
    return _mcp_parse(text)


def mcp_rpc(method, params=None):
    return _mcp_post({"jsonrpc": "2.0", "method": method, "params": params or {}})


def mcp_notify(method):
    _mcp_post({"jsonrpc": "2.0", "method": method, "params": {}}, is_notification=True)


def mcp_init():
    global _mcp_ready
    mcp_rpc("initialize", {
        "protocolVersion": "2025-11-25",
        "capabilities":    {},
        "clientInfo":      {"name": "fuse-fmz", "version": "1.0"},
    })
    mcp_notify("notifications/initialized")
    _mcp_ready = True
    Log("MCP ready  session_id=" + (_mcp_session_id or "(none)"))


def mcp_call_tool(name, arguments=None):
    result = mcp_rpc("tools/call", {"name": name, "arguments": arguments or {}})
    if "structuredContent" in result:
        return result["structuredContent"]
    for item in result.get("content", []):
        if item.get("type") != "text":
            continue
        try:
            return json.loads(item["text"])
        except json.JSONDecodeError:
            return {"raw": item["text"]}
    return result

# ─── News helpers ─────────────────────────────────────────────────────────────

def _extract_items(payload):
    if isinstance(payload, list):
        return [x for x in payload if isinstance(x, dict)]
    if not isinstance(payload, dict):
        return []
    for key in ("items", "data", "list", "rows", "flash", "news", "result"):
        v = payload.get(key)
        if isinstance(v, list):
            return [x for x in v if isinstance(x, dict)]
        if isinstance(v, dict):
            nested = _extract_items(v)
            if nested:
                return nested
    return []


def _pick(obj, *keys):
    for k in keys:
        v = obj.get(k)
        if v is not None and str(v).strip():
            return str(v).strip()
    return ""


import re

def _to_ms(ts_str):
    if not ts_str:
        return 0
    m = re.match(r"(\d{4}-\d{2}-\d{2})T(\d{2}:\d{2}:\d{2})([+-])(\d{2}):?(\d{2})?", ts_str.strip())
    if m:
        date_part, time_part, sign, tzh, tzm = m.groups()
        t = time.strptime(date_part + " " + time_part, "%Y-%m-%d %H:%M:%S")
        ts = calendar.timegm(t)
        offset = int(tzh) * 3600 + int(tzm or 0) * 60
        ts = ts - offset if sign == "+" else ts + offset
        return ts * 1000
    try:
        n = float(ts_str)
        return int(n * 1000 if n < 1e12 else n)
    except Exception:
        return 0


def _normalize(items, source):
    rows = []
    for it in items:
        title = _pick(it, "title", "content", "summary", "introduction", "name")
        if not title:
            continue
        ts_str = _pick(it, "time", "ts", "time_str", "datetime", "created_at", "display_time")
        full_text = " ".join(str(it.get(f, "")) for f in ("title", "content", "introduction", "summary"))
        rows.append({"ts": _to_ms(ts_str), "time": ts_str, "title": title, "source": source, "full_text": full_text})
    rows.sort(key=lambda x: x["ts"], reverse=True)
    return rows


def refresh_news():
    global _cached_news, _last_news_at, _mcp_ready
    if not JIN10_MCP_TOKEN:
        return
    now = int(time.time())
    if now - _last_news_at < NEWS_REFRESH_SEC and _cached_news:
        return
    _last_news_at = now
    try:
        if not _mcp_ready:
            mcp_init()
        flash_raw = mcp_call_tool("list_flash")
        news_raw  = mcp_call_tool("list_news")
        combined  = (
            _normalize(_extract_items(flash_raw), "flash") +
            _normalize(_extract_items(news_raw),  "news")
        )
        combined.sort(key=lambda x: x["ts"], reverse=True)
        _cached_news = combined[:80]
        Log("News updated: %d items" % len(_cached_news))
    except Exception as e:
        Log("News refresh failed: " + str(e))
        _mcp_ready = False

# ─── Symbol helpers ───────────────────────────────────────────────────────────

def switch_symbol(symbol):
    global _last_symbol
    if _last_symbol == symbol:
        return
    parts = symbol.split(".")
    exchange.SetCurrency(parts[0])
    exchange.SetContractType(parts[1] if len(parts) > 1 else "swap")
    _last_symbol = symbol


def get_records(symbol):
    switch_symbol(symbol)
    recs = exchange.GetRecords(symbol, PERIOD_MAP.get(KLINE_PERIOD, PERIOD_M1), KLINE_LIMIT)
    if not recs:
        return []
    recs = list(recs[-KLINE_LIMIT:])
    for r in recs:
        if r['Time'] < 1e12:
            r['Time'] = int(r['Time'] * 1000)
    return recs


def get_ticker(symbol):
    switch_symbol(symbol)
    return exchange.GetTicker(symbol)


def get_positions(symbol):
    switch_symbol(symbol)
    pos = exchange.GetPositions(symbol)
    return list(pos) if pos else []


def get_equity():
    try:
        acc = exchange.GetAccount()
        if not acc:
            return None
        return float(acc.Equity)
    except Exception as e:
        Log("get_equity error: " + str(e))
        return None

# ─── Equity tracking ──────────────────────────────────────────────────────────

def init_equity():
    global _init_equity
    saved = _G("init_equity")
    if saved is not None:
        _init_equity = float(saved)
        Log("Restored init_equity: %.4f" % _init_equity)
        return
    eq = get_equity()
    if eq is not None:
        _init_equity = eq
        _G("init_equity", eq)
        Log("Init equity recorded: %.4f" % eq)


def record_profit(equity):
    if _init_equity is None or equity is None:
        return
    LogProfit(round(equity - _init_equity, 4), '&')

# ─── Trade helpers ────────────────────────────────────────────────────────────

def market_order(symbol, action, amount):
    switch_symbol(symbol)
    qty = float(amount or _cur_amount)
    dir_map = {
        "openLong":   ("buy",       "open long"),
        "openShort":  ("sell",      "open short"),
        "closeLong":  ("closebuy",  "close long"),
        "closeShort": ("closesell", "close short"),
    }
    if action not in dir_map:
        return
    direction, memo = dir_map[action]
    Log("Trade: %s %s %s" % (action, qty, symbol))
    exchange.CreateOrder(symbol, direction, -1, qty, memo)


def close_all(symbol):
    for p in get_positions(symbol):
        qty = abs(float(p.Amount))
        if qty <= 0:
            continue
        direction = "closesell" if p.Type == PD_SHORT else "closebuy"
        Log("CloseAll: %s %s" % (direction, qty))
        exchange.CreateOrder(symbol, direction, -1, qty, "close all")

# ─── Command handler ──────────────────────────────────────────────────────────

def handle_command(symbol):
    global _cur_amount, _last_news_at
    cmd = GetCommand()
    if not cmd:
        return
    Log("CMD: " + cmd)
    parts = cmd.split(":")
    key   = parts[0]
    val   = parts[1] if len(parts) > 1 else ""
    if   key == "openLong":    market_order(symbol, "openLong",   _cur_amount)
    elif key == "openShort":   market_order(symbol, "openShort",  _cur_amount)
    elif key == "closeLong":   market_order(symbol, "closeLong",  _cur_amount)
    elif key == "closeShort":  market_order(symbol, "closeShort", _cur_amount)
    elif key == "closeAll":    close_all(symbol)
    elif key == "amount":
        _cur_amount = float(val)
        Log("Amount updated: " + str(_cur_amount))

# ─── Chart ────────────────────────────────────────────────────────────────────

def init_chart(symbol):
    global _chart
    _chart = Chart({
        "__isStock": True,
        "chart":     {"style": {"fontFamily": "Microsoft YaHei, SimHei, Arial, sans-serif"}},
        "title":     {"text": "FUSE  " + symbol},
        "xAxis":     {"type": "datetime"},
        "series": [
            {
                "id":   "kline",
                "type": "candlestick",
                "name": symbol,
                "data": [],
            },
            {
                "type":      "flags",
                "name":      "News",
                "onSeries":  "kline",
                "shape":     "circlepin",
                "color":     "#F59E0B",
                "fillColor": "#F59E0B",
                "width":     16,
                "data":      [],
            },
        ],
    })
    _chart.reset()


_last_bar_time   = 0
_last_news_hash  = 0   # 新闻变化时触发图表重置
_flagged_news_ts = set()  # 已绘制过的新闻 ts，避免重复 add

def draw_chart(records):
    global _last_bar_time, _last_news_hash, _flagged_news_ts

    if not _chart or not records:
        return

    # 检测新闻是否有更新，有则重置图表重画
    news_hash = hash(tuple(n.get("ts", 0) for n in _cached_news[:10]))
    news_changed = (news_hash != _last_news_hash)
    if news_changed:
        _chart.reset()
        _last_bar_time   = 0
        _last_news_hash  = news_hash
        _flagged_news_ts = set()

    # series 0：K线，增量 add
    for r in records:
        t   = r['Time']
        bar = [t, r['Open'], r['High'], r['Low'], r['Close']]
        if t > _last_bar_time:
            _chart.add(0, bar)
            _last_bar_time = t
        elif t == _last_bar_time:
            _chart.add(0, bar, -1)

    # series 1：关键词新闻 flag，对应到K线 bar 时间
    if not _cached_news:
        return
    kws = [k.strip() for k in NEWS_KEYWORD.split("|") if k.strip()]
    kw_news = [n for n in _cached_news if not kws or any(k in n.get("full_text", n["title"]) for k in kws)]
    if not kw_news:
        return
    p_ms  = PERIOD_MS.get(KLINE_PERIOD, 60000)
    first = records[0]['Time']
    last  = records[-1]['Time']
    by_bar = {}
    for n in kw_news:
        if not n.get("ts"):
            continue
        key = (n["ts"] // p_ms) * p_ms
        if key not in by_bar:
            by_bar[key] = n

    for ts, item in sorted(by_bar.items()):
        if not (first <= ts <= last):
            continue
        if ts in _flagged_news_ts:
            continue
        _chart.add(1, {
            "x": ts,
            "title": "📰",
            "text": item["title"][:100],
        })
        _flagged_news_ts.add(ts)

# ─── Status panel ─────────────────────────────────────────────────────────────

def make_status(symbol, ticker, positions, equity):
    # 1. 行情
    quote_rows = [[
        symbol,
        ticker.Last if ticker else "-",
        ticker.High if ticker else "-",
        ticker.Low  if ticker else "-",
        _D(ticker.Time) if ticker and ticker.Time else _D(),
    ]]

    # 2. 账户权益
    if equity is not None and _init_equity is not None:
        pnl = round(equity - _init_equity, 4)
        roi = round(pnl / _init_equity * 100, 2) if _init_equity else 0
        acc_rows = [[round(_init_equity, 4), round(equity, 4), pnl, str(roi) + "%"]]
    else:
        acc_rows = [["-", "-", "-", "-"]]

    # 3. 持仓
    pos_rows = []
    if positions:
        for p in positions:
            side = "空 SHORT" if p.Type == PD_SHORT else "多 LONG"
            pnl  = "-"
            if ticker and ticker.Last and p.Price:
                raw = (
                    (p.Price - ticker.Last) * abs(p.Amount)
                    if p.Type == PD_SHORT
                    else (ticker.Last - p.Price) * abs(p.Amount)
                )
                pnl = round(raw, 4)
            pos_rows.append([side, round(p.Amount, 4), round(p.Price, 4), pnl])
    else:
        pos_rows.append(["无持仓 / Flat", "-", "-", "-"])

    # 4. 关键词新闻（显示命中的关键词，而不是来源）
    kws = [k.strip() for k in NEWS_KEYWORD.split("|") if k.strip()]
    kw_rows = []
    for item in _cached_news[:40]:
        t     = item.get("time") or (_D(item["ts"]) if item.get("ts") else "-")
        title = item["title"][:90]
        text  = item.get("full_text", item["title"])
        hit_kws = [k for k in kws if k in text]
        if hit_kws:
            kw_rows.append([t, "/".join(hit_kws), title])

    if not kw_rows:
        kw_rows = [["-", "-", "暂无关键词相关新闻"]]

    # 5. 实时新闻（最新5条）
    all_rows = []
    for item in _cached_news[:5]:
        t     = item.get("time") or (_D(item["ts"]) if item.get("ts") else "-")
        src   = item.get("source", "-")
        title = item["title"][:90]
        all_rows.append([t, src, title])
    if not all_rows:
        all_rows = [["-", "-", "新闻加载中..." if JIN10_MCP_TOKEN else "未配置 MCP Token"]]

    LogStatus(
        "`" + json.dumps({
            "type": "table", "title": "行情 Quotes",
            "cols": ["Symbol", "Last", "High", "Low", "Time"],
            "rows": quote_rows,
        }) + "`\n" +
        "`" + json.dumps({
            "type": "table", "title": "账户权益 Equity",
            "cols": ["初始权益 Init", "当前权益 Now", "盈亏 PnL", "收益率 ROI"],
            "rows": acc_rows,
        }) + "`\n" +
        "`" + json.dumps({
            "type": "table", "title": "持仓 Positions",
            "cols": ["方向 Side", "数量 Amount", "均价 AvgPrice", "浮盈 UPnL"],
            "rows": pos_rows,
        }) + "`\n" +
        "`" + json.dumps({
            "type": "table", "title": "关键词新闻 (" + NEWS_KEYWORD + ")",
            "cols": ["时间", "关键词", "标题"],
            "rows": kw_rows,
        }) + "`\n" +
        "`" + json.dumps({
            "type": "table", "title": "实时新闻 Live News (最新5条)",
            "cols": ["时间", "来源", "标题"],
            "rows": all_rows,
        }) + "`"
    )

# ─── Main ─────────────────────────────────────────────────────────────────────

def main():
    _G(None)
    if not SYMBOL:
        raise RuntimeError("SYMBOL is empty")

    # 先切换合约，确保 GetAccount 返回正确权益
    switch_symbol(SYMBOL)

    # 恢复或记录初始权益
    init_equity()

    # 初始化 MCP session
    if JIN10_MCP_TOKEN:
        try:
            mcp_init()
        except Exception as e:
            Log("MCP init failed: " + str(e))

    init_chart(SYMBOL)
    Log("FUSE started | %s | %s | amount=%s" % (SYMBOL, KLINE_PERIOD, _cur_amount))

    while True:
        handle_command(SYMBOL)
        refresh_news()

        records   = get_records(SYMBOL)
        ticker    = get_ticker(SYMBOL)
        positions = get_positions(SYMBOL)
        equity    = get_equity()

        record_profit(equity)

        if records:
            draw_chart(records)

        make_status(SYMBOL, ticker, positions, equity)

        Sleep(max(1, REFRESH_SECONDS) * 1000)