एक, प्रारंभ बिंदु: ट्रंप द्वारा हाथ से बनाई गई K-लाइन

समाचार की समयबद्धता निस्संदेह महत्वपूर्ण है। अमेरिका द्वारा ईरान पर हवाई हमले की खबर आने के बाद, कच्चे तेल की कीमतों में थोड़े समय में भारी उछाल आया, और इस प्रक्रिया में, ट्रंप और ईरान के बयान जैसे कारक लगातार आपस में जुड़ते और एक-दूसरे को मजबूत करते रहे, जिससे बाजार को नई-नई सीमाओं की ओर धकेल दिया गया।

हम अक्सर मजाक में कहते हैं कि "ट्रंप हाथ से K-लाइन बनाते हैं" - कई बार कीमतों में तेज उतार-चढ़ाव तकनीकी संकेतकों के विकास से नहीं, बल्कि एक ट्वीट, एक भाषण, एक नीतिगत बयान से सीधे उत्पन्न होता है। तकनीकी विश्लेषण आपको बताता है कि "अब आप कहाँ हैं", लेकिन अक्सर समाचार ही वह प्रमुख चर होता है जो बताता है कि "यह यहाँ क्यों है और आगे कहाँ जा सकता है"।

समाचार का महत्व निर्विवाद है, लेकिन समस्या यथार्थवादी भी है: एक इंसान 24 घंटे चार्ट और समाचार पर नज़र नहीं रख सकता, और सूचना के प्रवाह में पहली बार में उस खबर को पकड़ना और भी मुश्किल है जो वास्तव में बाजार को हिला देगी। तब एक सरल विचार उभरता है - क्या हम समाचार को सीधे K-लाइन चार्ट पर "चित्रित" कर सकते हैं, ताकि कीमत और संदेश एक ही दृश्य में एक साथ प्रस्तुत हो सकें? कम से कम "देखने" की समस्या को तो हल करें।

दो, समाचार स्रोत का चयन: MCP के माध्यम से जिनशी में प्रवेश

समाचारों को जोड़ने के लिए पहला कदम एक ऐसा स्रोत खोजना है जो पर्याप्त तेजी से अपडेट होता हो और जिसकी संरचना अपेक्षाकृत मानक हो। इस बार हमने जिनशी डेटा (Jin10 Data) का चयन किया, और इसे 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-लाइन बार पर संरेखित करें:

            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-लाइन, वृद्धिशील जोड़ें
    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: कीवर्ड समाचार फ़्लैग, K-लाइन बार के समय से संबंधित
    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-लाइन बार से जोड़ा जाता है, सामग्री स्तर पर कोई व्याख्या नहीं की जाती। कोई खबर सकारात्मक है या नकारात्मक, और क्या वह बाजार में हलचल लाएगी, यह पूरी तरह से मनुष्य पर निर्भर करता है।

दूसरी, कीवर्ड फ़िल्टरिंग अपेक्षाकृत सरल समाधान है। कीवर्ड मैच का मतलब यह नहीं कि खबर वास्तव में महत्वपूर्ण है, और न मैच का मतलब यह नहीं कि वह महत्वपूर्ण नहीं है – इस बीच व्यक्तिगत ज्ञान, अनुभव, और यहाँ तक कि उस दिन की मानसिक स्थिति का बहुत बड़ा भार हो सकता है – एक ही खबर, अलग-अलग लोग एक ही चार्ट पर देखकर पूरी तरह से अलग निष्कर्ष निकाल सकते हैं।

तीसरी, पूरी प्रक्रिया अब भी "मानव-इन-द-लूप" है, प्रतिक्रिया की गति मनुष्य की प्रतिक्रिया गति से सीमित है, जबकि कई बार बाजार की प्रतिक्रिया समाचार पर मिनटों या सेकंडों में होती है।

यदि आप सभी को इस दिशा में रुचि है, तो आगे हम एक बड़े मॉडल-आधारित स्वचालित संस्करण बनाने का प्रयास करेंगे, जहाँ मॉडल समाचारों की प्रारंभिक व्याख्या और महत्व निर्धारण करेगा, मानव निर्णय के सहायक या विकल्प के रूप में। यदि रुचि है, तो कृपया अनुसरण करते रहें।

रणनीति स्रोत कोड: リアルタイムニュース導火線システム