I. Titik Awal: Trump Menggambar K Line Sendiri

Ketepatan waktu berita tidak perlu diragukan lagi. Setelah berita serangan udara AS terhadap Iran menyebar, harga minyak mentah melonjak tajam dalam waktu singkat. Dalam proses ini, pernyataan Trump, Iran, dan faktor lainnya terus saling terkait dan saling memperkuat, mendorong pasar ke level-level baru.

Kita sering bercanda "Trump menggambar K Line sendiri" — seringkali fluktuasi harga yang tajam bukan berasal dari evolusi indikator teknis, melainkan langsung dihasilkan oleh sebuah tweet, sebuah pidato, atau sebuah pernyataan kebijakan. Analisis teknis memberi tahu Anda "di mana sekarang", tetapi berita seringkali menjadi variabel kunci untuk "mengapa di sini, dan ke mana selanjutnya akan bergerak".

Pentingnya berita tidak perlu diragukan, tetapi masalahnya juga realistis: manusia tidak bisa memantau layar dan berita 24 jam, apalagi dalam banjir informasi langsung menangkap berita yang benar-benar akan memicu pergerakan pasar. Maka muncul ide sederhana — bisakah kita langsung "menggambar" berita pada grafik K line, sehingga harga dan berita ditampilkan bersamaan dalam satu tampilan? Setidaknya selesaikan dulu masalah "melihat".

II. Pemilihan Sumber Berita: Akses Jin10 melalui MCP

Untuk menghubungkan berita, langkah pertama adalah mencari sumber berita yang cukup cepat diperbarui dan strukturnya relatif standar. Kali ini kami menggunakan data Jin10, terhubung melalui MCP (Model Context Protocol), memanggil dua jenis antarmuka: list_flash (flash news) dan list_news (berita).

Saya tidak akan membahas terlalu banyak tentang Jin10 itu sendiri — itu hanya pilihan kami saat ini, idenya terlepas dari sumber berita spesifik. Selama suatu sumber dapat menyediakan judul/teks dengan stempel waktu dan dapat dipanggil dengan cara standar seperti MCP, sumber tersebut bisa diganti. Yang penting adalah desain lapisan koneksi ini, bukan mengikat pada alat tertentu.

Koneksi MCP dan manajemen sesi adalah bagian yang cukup "mendasar" namun sangat penting dalam sistem ini:

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

Setelah sesi terbentuk, mengambil berita hanya membutuhkan dua panggilan alat, lalu dilakukan normalisasi format dan deduplikasi:

            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

Nama bidang yang dikembalikan oleh sumber berita berbeda sangat bervariasi (title/content/introduction, time/ts/created_at ...), jadi kami menambahkan lapisan _extract_items + _normalize untuk menyatukan semua format menjadi struktur standar {ts, time, title, source, full_text}, sehingga logika grafik dan penyaringan selanjutnya tidak perlu peduli dari antarmuka mana data berasal.

Catatan: Anda perlu mengajukan aplikasi untuk menggunakan API MCP.

III. Mekanisme Inti: Membuat Berita "Tumbuh" di K Line

Berikutnya adalah bagian yang benar-benar menarik dari alat ini — menempatkan berita dan K line dalam satu grafik yang sama.

Kami menambahkan series kedua pada grafik, tipe flags, yang ditempelkan pada series K line sebagai "lapisan penanda berita":

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

Setiap kali refresh, pertama perbarui data candlestick secara inkremental, lalu filter berita "penting" berdasarkan kata kunci, sejajarkan berita tersebut dengan bar candlestick yang sesuai berdasarkan waktu:

            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)

Hasilnya: setiap kali muncul tanda 📰 pada grafik, ketika mouse diarahkan ke atasnya, akan terlihat judul berita yang sesuai, dan posisinya tepat pada candlestick yang sesuai dengan waktu terjadinya berita tersebut. Titik balik harga dan titik waktu berita, untuk pertama kalinya, disajikan dalam satu tampilan secara intuitif — Anda tidak perlu lagi beralih antara dua jendela untuk melihat "bagian ini karena apa".

NEWS_KEYWORD mendukung pemisahan beberapa kata kunci dengan | (misalnya "Iran|kenaikan suku bunga|nonfarm|tarif"), sistem akan memprioritaskan untuk menandai berita yang cocok dengan kata kunci pada grafik, menghindari grafik dibanjiri berita singkat yang tidak relevan.

Empat、Panel Status: Pergerakan Harga, Posisi, Berita dalam Satu Layar

Selain grafik, kami juga membuat sekelompok tabel status, yang ditampilkan melalui LogStatus, termasuk pergerakan harga real-time, ekuitas akun dan laba/rugi, posisi saat ini, berita yang cocok dengan kata kunci, serta berita singkat lengkap terbaru:

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

Ditambah dengan antarmuka perintah manual sederhana — buka long, buka short, tutup long, tutup short, satu tombol untuk menutup semua, dan ubah jumlah pesanan, semuanya diterima melalui 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))

Secara keseluruhan, FUSE pada dasarnya adalah panel pemantauan "integrasi informasi + eksekusi manual": ia menampilkan harga, berita, posisi, dan status akun sebisa mungkin dalam satu layar. Pengambilan keputusan tetap sepenuhnya ada pada manusia—ia tidak menggantikan penilaian Anda, tetapi membantu Anda agar tidak melewatkan terlalu banyak hal saat mengambil keputusan.

V. Keterbatasan: Manusia, Tetap Variabel Terbesar

Keterbatasan versi ini sebenarnya cukup jelas, dan kami tidak ingin menghindarinya.

Pertama, hubungan antara berita dan harga bersifat "kasar"—berita hanya ditempelkan ke bar K-line yang sesuai berdasarkan stempel waktu, tanpa interpretasi pada level konten. Apakah suatu berita bersifat positif atau negatif, apakah akan memicu pergerakan pasar, sepenuhnya bergantung pada penilaian manusia.

Kedua, penyaringan kata kunci itu sendiri adalah solusi yang cukup sederhana. Kata kunci yang terdeteksi belum tentu berarti berita itu penting, dan yang tidak terdeteksi belum tentu tidak penting. Di sini, kognisi, pengalaman, bahkan kondisi hari itu dari masing-masing individu dapat memegang bobot yang besar—berita yang sama, dilihat oleh orang yang berbeda di bagan yang sama, dapat menghasilkan kesimpulan yang sangat berbeda.

Ketiga, seluruh proses masih "manusia dalam putaran", kecepatan respons terbatas oleh kecepatan reaksi manusia, dan seringkali reaksi pasar terhadap berita terjadi dalam hitungan menit atau bahkan detik.

Jika kalian tertarik dengan arah ini, ke depannya kami akan mencoba membuat versi otomatis berbasis model besar, yang memungkinkan model melakukan interpretasi awal dan penilaian kepentingan berita, sebagai bantuan atau bahkan pengganti keputusan manual. Jika berminat, silakan terus pantau.

Kode sumber strategi: Sistem Pemicu Berita Real-time