1. Ponto de partida: Trump desenha o gráfico de velas

A atualidade das notícias é inquestionável. Após a notícia do ataque aéreo dos EUA ao Irã, o preço do petróleo disparou violentamente em um curto período. Nesse processo, fatores como as declarações de Trump e do Irã continuaram se entrelaçando e reforçando mutuamente, empurrando o mercado para novas faixas de preço uma após outra.

Costumamos brincar dizendo "Trump desenha o gráfico de velas" — muitas vezes, as oscilações bruscas de preços não são geradas por indicadores técnicos, mas sim por um tweet, um discurso, uma declaração política que impacta diretamente o mercado. A análise técnica diz "onde estamos agora", mas são as notícias que geralmente revelam "por que estamos aqui e para onde podemos ir a seguir".

A importância das notícias é inquestionável, mas o problema é realista: é impossível para uma pessoa monitorar o mercado e as notícias 24 horas por dia, e ainda mais difícil capturar em meio ao fluxo de informações justamente aquela mensagem que realmente desencadeará movimentos. Surge então uma ideia simples — será possível "desenhar" as notícias diretamente no gráfico de velas, de modo que preço e informação apareçam simultaneamente na mesma visualização? Pelo menos resolver o problema de "ver".

2. Escolha da fonte de notícias: integração com Jin10 via MCP

Para integrar as notícias, o primeiro passo é encontrar uma fonte que seja atualizada rapidamente e tenha uma estrutura relativamente padronizada. Desta vez, optamos pelo Jin10 Data, acessando-o através do MCP (Model Context Protocol), utilizando as interfaces list_flash (flash) e list_news (notícias).

Não pretendo me alongar muito sobre o Jin10 em si — é apenas uma escolha atual; a lógica é desacoplada da fonte específica. Desde que uma fonte forneça títulos/textos com timestamp e possa ser chamada de forma padronizada via MCP, pode ser substituída. O foco está no design dessa camada de integração, não em uma ferramenta específica.

A conexão e o gerenciamento de sessão do MCP são partes relativamente "de baixo nível", mas cruciais neste sistema:

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

Depois que a sessão é estabelecida, obter as notícias se resume a duas chamadas de ferramenta, seguidas de normalização e deduplicação uniformes:

            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

Fontes de notícias diferentes usam nomes de campos variados (title/content/introduction, time/ts/created_at...). Por isso, foi adicionada uma camada intermediária _extract_items + _normalize, que unifica todos os formatos em uma estrutura padrão {ts, time, title, source, full_text}. Assim, a lógica de gráficos e filtros posteriores não precisa saber de qual interface os dados vieram.

Atenção: É necessário solicitar a API MCP para uso.

3. Mecânica principal: fazer com que as notícias "cresçam" no gráfico de velas

Agora vem a parte realmente interessante dessa ferramenta — colocar as notícias e o gráfico de velas na mesma figura.

Adicionamos uma segunda série ao gráfico, com tipo flags, ancorada na série de velas, funcionando como uma "camada de marcadores de notícias":

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

A cada atualização, primeiro atualize incrementalmente os dados de velas, depois filtre as notícias "importantes" com base nas palavras-chave e alinhe-as ao bar da vela correspondente no tempo:

            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)

O efeito é: cada vez que um marcador 📰 aparece no gráfico, ao passar o mouse sobre ele, você vê o título da notícia correspondente, e sua posição é exatamente a vela correspondente ao momento em que a notícia ocorreu. Os pontos de inflexão de preço e os momentos das notícias são colocados no mesmo quadro de forma intuitiva pela primeira vez — você não precisa mais alternar entre duas janelas para entender "o que causou esta seção".

NEWS_KEYWORD suporta a separação de múltiplas palavras-chave por | (por exemplo, "Irã|aumento de juros|não agrícola|tarifas"), e o sistema priorizará marcar no gráfico as notícias que correspondem às palavras-chave, evitando que o gráfico seja poluído por notícias irrelevantes.

IV. Painel de status: cotações, posições e notícias em uma tela

Além do gráfico, criamos um conjunto de tabelas de status, que são exibidas através de LogStatus, incluindo cotações em tempo real, patrimônio da conta e lucros/perdas, posições atuais, notícias que correspondem às palavras-chave e as últimas notícias completas:

            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 = [["-", "-", "暂无关键词相关新闻"]]

Além disso, há uma interface simples de comandos manuais — abrir longo, abrir curto, fechar longo, fechar curto, fechar tudo com um clique, modificar a quantidade de ordens, todos recebidos através de 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))

No conjunto, o FUSE é essencialmente um "painel de monitoramento que integra informações e execução manual": ele coloca preços, notícias, posições e status da conta na mesma tela o máximo possível, mas a decisão continua totalmente com o ser humano — ele não decide por você, mas tenta garantir que você não perca nada ao tomar decisões.

V. Limitações: O ser humano ainda é a maior variável

As limitações desta versão são bastante evidentes, e não vamos evitá-las.

Primeiro, a correspondência entre notícias e preços é "grosseira" — as notícias são apenas anexadas às barras de candle correspondentes com base no timestamp, sem qualquer interpretação de conteúdo. Se uma notícia é positiva ou negativa, e se irá desencadear movimentos de preço, depende totalmente do julgamento humano.

Segundo, a filtragem por palavras-chave é uma abordagem relativamente simples. A presença de uma palavra-chave não significa que a notícia seja realmente importante, e a ausência também não significa que não seja. Nesse processo, a cognição pessoal, a experiência e até mesmo o estado de espírito do dia podem ter um peso enorme — a mesma notícia, observada por pessoas diferentes no mesmo gráfico, pode levar a conclusões completamente distintas.

Terceiro, todo o fluxo ainda é "humano no circuito", e a velocidade de resposta é limitada pela reação humana, enquanto muitas vezes o mercado reage às notícias em minutos ou até segundos.

Se vocês se interessam por essa direção, no futuro tentaremos criar uma versão automatizada baseada em grandes modelos, onde o modelo fará a interpretação preliminar e a avaliação de importância das notícias, servindo como auxílio ou até substituto para a decisão humana. Se houver interesse, sintam-se à vontade para continuar acompanhando.

Código-fonte da estratégia: Sistema de Gatilho de Notícias em Tempo Real