I. Point de départ : le K-line dessiné par Trump

L'actualité a indéniablement un fort impact. Après l'annonce du raid aérien américain contre l'Iran, les prix du pétrole brut ont grimpé brusquement en très peu de temps. Au cours de ce processus, les déclarations de Trump et de l'Iran se sont entremêlées et renforcées mutuellement, poussant les cours vers de nouvelles zones.

Nous plaisantons souvent en disant que "Trump dessine les bougies" — bien souvent, les fluctuations brutales des prix ne sont pas le fruit d'indicateurs techniques, mais sont directement provoquées par un tweet, un discours, une prise de position politique. L'analyse technique vous dit "où vous êtes", mais ce sont souvent les actualités qui constituent la variable clé pour comprendre "pourquoi on est ici et où on pourrait aller ensuite".

L'importance des actualités ne fait aucun doute, mais le problème est concret : personne ne peut surveiller les marchés et les nouvelles 24 heures sur 24, et il est encore plus difficile, dans le flot d'informations, d'attraper en premier le message qui déclenchera réellement un mouvement de marché. De là naît une idée très simple : peut-on "dessiner" directement les nouvelles sur le graphique en bougies, afin que les prix et les informations soient présentés simultanément dans la même vue ? Au moins, cela permettrait de résoudre d'abord le problème de la "visualisation".

II. Choix de la source d'actualités : intégration de Jin10 via MCP

Pour intégrer les actualités, la première étape est de trouver une source suffisamment rapide et dont la structure est relativement standard. Cette fois, nous avons choisi Jin10 Data (金十数据), en l'intégrant via le protocole MCP (Model Context Protocol), en appelant deux types d'interfaces : list_flash (flash info) et list_news (actualités).

Nous ne détaillerons pas trop Jin10 ici — ce n'est qu'un choix actuel. L'idée générale est découplée de la source spécifique : tant qu'une source peut fournir un titre/texte avec un timestamp et être appelée de manière standardisée via MCP, elle peut être substituée. L'important est la conception de cette couche d'intégration, pas la liaison à un outil particulier.

La connexion et la gestion de session de MCP constituent une partie plutôt "bas niveau" mais cruciale de ce système :

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

Une fois la session établie, récupérer les actualités se résume à deux appels d'outils, suivis d'une normalisation et d'un dédoublonnage du format :

            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

Les noms de champs renvoyés par différentes sources sont très variables (title/content/introduction, time/ts/created_at...). C'est pourquoi nous avons ajouté une couche intermédiaire _extract_items + _normalize pour uniformiser tous les formats en une structure standard {ts, time, title, source, full_text}. Ainsi, la logique de graphique et de filtrage n'a pas à se soucier de la provenance des données.

Note : Une clé API MCP est nécessaire pour utiliser cette fonctionnalité.

III. Fonctionnalité clé : faire "pousser" les actualités sur les bougies

Voici la partie vraiment intéressante de cet outil — placer les actualités et les bougies sur le même graphique.

Nous avons ajouté une deuxième série dans le graphique, de type flags, liée à la série des bougies, servant de "couche de marqueurs d'actualités" :

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

Lors de chaque rafraîchissement, mettez d'abord à jour les données de chandelier de manière incrémentielle, puis filtrez les nouvelles "importantes" en fonction des mots clés, et alignez-les temporellement sur la barre de chandelier correspondante :

            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)

Le résultat est le suivant : chaque fois qu'un marqueur 📰 apparaît sur le graphique, passer la souris dessus affiche le titre de la nouvelle correspondante, et son emplacement est exactement la barre de chandelier correspondant au moment de l'événement. Les points d'inflexion des prix et les points temporels des nouvelles sont désormais placés dans une même vue intuitive – vous n'avez plus besoin de basculer entre deux fenêtres pour comprendre « à quoi est dû ce mouvement ».

NEWS_KEYWORD prend en charge la séparation de plusieurs mots clés par | (par exemple « Iran|hausse des taux|non-agricole|droits de douane »). Le système donne la priorité au marquage des nouvelles contenant ces mots clés sur le graphique, évitant ainsi que le graphique ne soit submergé par des dépêches non pertinentes.

IV. Panneau d'état : cours, positions et nouvelles en un seul coup d'œil

En plus du graphique, nous avons créé un ensemble de tableaux d'état, affichés via LogStatus, incluant les cotations en temps réel, les capitaux propres et les gains/pertes du compte, les positions actuelles, les nouvelles correspondant aux mots clés, ainsi que les dernières dépêches complètes :

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

De plus, une interface simple de commandes manuelles – ouvrir long, ouvrir court, fermer long, fermer court, tout fermer d'un coup, modifier la quantité de l'ordre – toutes sont reçues via 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))

Dans l'ensemble, FUSE est fondamentalement un tableau de bord de surveillance qui combine l'intégration d'informations et l'exécution manuelle : il rassemble les prix, les actualités, les positions et l'état du compte sur un seul écran. La décision reste entièrement humaine – il ne juge pas à votre place, mais il vous permet de ne rien oublier lorsque vous prenez une décision.

V. Limites : l'humain reste la variable la plus importante

Les limites de cette version sont assez évidentes, et nous ne cherchons pas à les cacher.

Premièrement, la correspondance entre les actualités et les prix est « grossière » – les actualités sont simplement associées à la bougie correspondante selon leur horodatage, sans interprétation du contenu. Savoir si une actualité est positive ou négative, et si elle va déclencher un mouvement, repose entièrement sur le jugement humain.

Deuxièmement, le filtrage par mots-clés est une approche assez basique. Un mot-clé correspondant ne garantit pas que l'actualité soit vraiment importante, et l'absence de correspondance ne signifie pas qu'elle ne l'est pas. La cognition, l'expérience, voire l'état d'esprit du jour de l'utilisateur peuvent peser lourd – face à la même actualité sur le même graphique, deux personnes peuvent tirer des conclusions totalement différentes.

Troisièmement, le processus reste « en boucle humaine », la vitesse de réponse est limitée par le temps de réaction de l'humain, alors que bien souvent la réaction du marché à une actualité se joue en minutes, voire en secondes.

Si ce sujet vous intéresse, nous pourrions tenter par la suite une version automatisée basée sur un grand modèle de langage, permettant au modèle de faire une première lecture et une évaluation de l'importance des actualités, en tant qu'aide – voire remplacement – de la décision humaine. Si cela vous intéresse, n'hésitez pas à nous suivre.

Code source de la stratégie : Système de déclenchement en temps réel basé sur les actualités