1. Начало: K-линии, нарисованные рукой Трампа

Актуальность новостей не вызывает сомнений. После сообщения об авиаударе США по Ирану цены на нефть резко взлетели за короткое время. В этом процессе заявления Трампа, Ирана и других факторов постоянно переплетались и взаимно усиливали друг друга, толкая рынок к новым уровням.

Мы часто шутим, что «Трамп рисует K-линии своей рукой» – зачастую резкие колебания цен возникают не из-за изменения технических индикаторов, а из-за одного твита, выступления или политического заявления. Технический анализ говорит вам «где вы сейчас», но именно новости часто являются ключевым фактором, объясняющим «почему вы здесь и куда можно двинуться дальше».

Важность новостей неоспорима, но проблема очевидна: человек не может 24 часа в сутки следить за графиком и новостями, а в потоке информации сложно сразу уловить ту новость, которая действительно вызовет движение рынка. Поэтому возникла простая идея – можно ли «нанести» новости прямо на K-линии, чтобы цена и сообщения отображались одновременно на одном графике? Хотя бы чтобы решить проблему «видимости».

2. Выбор источника новостей: подключение Jin10 через MCP

Чтобы подключить новости, первый шаг – найти источник с достаточно быстрым обновлением и относительно стандартной структурой. В этот раз мы выбрали Jin10 Data (金十数据) и подключаемся через MCP (Model Context Protocol), используя два типа интерфейсов: list_flash (мгновенные новости) и list_news (новости).

Здесь не будем тратить много времени на описание самого Jin10 – это лишь наш текущий выбор. Идея и конкретный источник новостей не связаны: если какой-то источник может предоставлять заголовки/текст с временными метками и вызываться стандартным способом через 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}. После этого логика отрисовки графиков и фильтрации уже не зависит от того, из какого интерфейса пришли данные.

Примечание: Для использования требуется получить API-ключ MCP.

3. Основная идея: «привязать» новости к K-линиям

Теперь самое интересное – поместить новости и K-линии на один график.

Мы добавили вторую серию на график, тип flags, которая накладывается на серию K-линий и служит «слоем меток новостей»:

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

При каждом обновлении сначала инкрементально обновляются данные свечных графиков, затем по ключевым словам фильтруются «важные» новости и привязываются по времени к соответствующим барам свечного графика:

            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: свечной график, инкрементальное добавление
    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: флаги новостей по ключевым словам, привязка ко времени бара свечного графика
    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)

Результат: на графике каждый раз, когда появляется метка 📰, при наведении мыши отображается соответствующий заголовок новости, причём её положение соответствует бару свечного графика, на момент которого произошла эта новость. Точки разворота цен и моменты новостей впервые оказываются на одном экране в наглядном виде — больше не нужно переключаться между двумя окнами, чтобы понять, «из-за чего произошёл этот участок».

NEWS_KEYWORD поддерживает разделение нескольких ключевых слов через | (например, "Иран|повышение ставки|Nonfarm|тарифы"); система в первую очередь помечает на графике новости, содержащие ключевые слова, чтобы избежать засорения графика нерелевантными сообщениями.

4. Панель состояния: котировки, позиции, новости — всё на одном экране

Помимо графика, мы также создали набор таблиц состояния, выводимых через 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 по сути представляет собой панель для наблюдения за рынком, объединяющую "интеграцию информации + ручное выполнение": она старается разместить цены, новости, позиции и состояние счета на одном экране. Решение полностью остается за человеком — она не делает выводы за вас, но помогает не упустить важные детали при принятии решений.

V. Ограничения: человек – по-прежнему самая большая переменная

Ограничения этой версии вполне очевидны, и мы не собираемся их скрывать.

Первое. Соответствие между новостями и ценами носит "грубый" характер – новости просто привязываются по временной метке к соответствующей свече на графике, без анализа содержания. Является ли конкретная новость позитивной или негативной, вызовет ли она движение рынка – полностью определяется человеком.

Второе. Фильтрация по ключевым словам – это довольно примитивное решение. Совпадение с ключевым словом не означает, что новость действительно важна, а отсутствие совпадения не означает неважности. В этом случае личное восприятие, опыт и даже текущее настроение могут играть большую роль – один и тот же график с одной и той же новостью может привести к совершенно разным выводам у разных людей.

Третье. Весь процесс по-прежнему строится по схеме "человек в контуре", скорость реакции ограничена человеческой реакцией, а в большинстве случаев рынок реагирует на новости в течение минут или даже секунд.

Если вам интересно это направление, в будущем мы попробуем создать автоматизированную версию на основе большой языковой модели, которая будет выполнять предварительный анализ новостей и оценку их значимости, выступая в качестве помощника или даже замены для принятия решений человеком. Если интересно – следите за обновлениями.

Исходный код стратегии: Система "Запал" для торговли по новостям в реальном времени