一、起点：特朗普手画K线

新闻的时效性毋庸置疑。美国空袭伊朗的消息传出后，原油价格在短时间内剧烈拉升，而在这一过程中，特朗普和伊朗的表态等因素持续交织、互相强化，把行情推向一个又一个新的区间。

我们常开玩笑说"特朗普手画K线"——很多时候价格的剧烈波动根本不是技术指标演变出来的，而是一条推文、一场讲话、一次政策表态直接砸出来的。技术分析告诉你"现在在哪"，但新闻往往才是"为什么会在这里，接下来可能往哪走"的关键变量。

新闻的重要性毋庸置疑，但问题也很现实：人不可能24小时盯盘盯新闻，更难在信息洪流里第一时间抓住那条真正会引发行情的消息。于是一个很朴素的想法冒出来——能不能把新闻直接"画"在K线图上，让价格和消息在同一个视图里同时呈现？至少先解决"看见"的问题。

二、新闻源的选择：MCP 接入金十

要把新闻接进来，第一步是找一个更新够快、结构相对标准的新闻源。这次我们选用了金十数据，通过 MCP（Model Context Protocol）的方式接入，调用 list_flash（快讯）和 list_news（资讯）两类接口。

这里不打算花太多篇幅介绍金十本身——它只是我们目前用的一个选择，思路和具体新闻源是解耦的，只要某个源能提供带时间戳的标题/正文、并且能用类似 MCP 的标准方式调用，都可以替换进来。重点是接入这一层的设计，而不是绑定某个特定工具。

MCP 的连接和会话管理是这套系统里比较"底层"但也很关键的一块：

            python
            
            
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_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)"))

会话建立之后，拉新闻就是两次工具调用，再做统一的格式归一化和去重：

            python
            
            
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

不同新闻源返回的字段名千奇百怪（title/content/introduction、time/ts/created_at……），所以中间加了一层 _extract_items + _normalize，把各种格式都统一成 {ts, time, title, source, full_text} 的标准结构，后面的图表和过滤逻辑就不用关心数据到底来自哪个接口。

注意： 需要申请MCP的API方可使用。

三、核心玩法：让新闻"长"在K线上

接下来是这个工具真正有意思的部分——把新闻和K线放在同一张图里。

我们在图表里加了第二条 series，类型是 flags，挂在K线 series 上，作为"新闻标记层"：

            python
            
            
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()

每次刷新时，先增量更新K线数据，再根据关键词过滤出"重要"新闻，把它们按时间对齐到对应的K线 bar 上：

            python
            
            
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)

效果就是：图上每出现一个📰标记，鼠标移上去就能看到对应的新闻标题，而它所在的位置正是那条新闻发生时间对应的K线。价格的拐点和新闻的时间点，第一次以一种直观的方式摆在了同一个画面里——你不再需要在两个窗口之间切换着看"这一段是因为什么"。

NEWS_KEYWORD 支持用 | 分隔多个关键词（比如 "伊朗|加息|非农|关税"），系统会优先把命中关键词的新闻标到图上，避免图表被无关快讯刷屏。

四、状态面板：行情、持仓、新闻一屏看完

除了图表，我们还做了一组状态表格，通过 LogStatus 输出，包括实时行情、账户权益与盈亏、当前持仓、命中关键词的新闻、以及最新的全量快讯：

            python
            
            
def make_status(symbol, ticker, positions, equity):
    # 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 = [["-", "-", "暂无关键词相关新闻"]]

再加上一个简单的手动指令接口——开多、开空、平多、平空、一键全平、修改下单数量，全部通过 GetCommand() 接收：

            python
            
            
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))

整套下来，FUSE 本质上是一个"信息整合 + 手动执行"的看盘面板：它把价格、新闻、持仓、账户状况尽量放在同一屏幕里，决策依然完全在人——它不替你判断，但尽量让你判断时少漏看一些东西。

五、局限性：人，仍然是最大的变量

这个版本的局限其实相当明显，我们也不想回避。

第一，新闻和价格的对应关系是"粗粒度"的——只是把新闻按时间戳挂到对应的K线 bar 上，并没有做内容层面的解读。一条新闻到底是利好还是利空、会不会引发行情，完全靠人来判断。

第二，关键词过滤本身是个比较朴素的方案。命中关键词不代表新闻真的重要，没命中也不代表不重要，这中间个人的认知、经验，甚至当天的状态，可能会占很大的权重——同一条新闻，不同人盯着同一张图，可能得出完全不同的结论。

第三，整个流程仍然是"人在回路"的，响应速度受限于人的反应速度，而很多时候行情对新闻的反应是分钟级甚至秒级的。

如果大家对这个方向感兴趣，我们后续会尝试做一个基于大模型的自动化版本，让模型来做新闻的初步解读和重要性判断，作为人工决策的辅助甚至替代。有兴趣的话，欢迎持续关注。

策略源码： 实时新闻导火线系统