১. শুরু: ট্রাম্পের হাতে আঁকা K-লাইন
খবরের সময়োপযোগিতা নিঃসন্দেহে গুরুত্বপূর্ণ। ইরানে মার্কিন বিমান হামলার খবর ছড়িয়ে পড়ার পর অপরিশোধিত তেলের দাম স্বল্প সময়ের মধ্যে তীব্রভাবে বেড়ে যায় এবং এই প্রক্রিয়ায় ট্রাম্প ও ইরানের বক্তব্য প্রভৃতি বিষয়গুলো ক্রমাগতভাবে একে অপরকে প্রভাবিত করে এবং শক্তিশালী করে, দামকে নতুন নতুন পর্যায়ে নিয়ে যায়।
আমরা প্রায়ই ঠাট্টা করে বলি "ট্রাম্পের হাতে আঁকা K-লাইন" – অনেক সময় দামের তীব্র ওঠানামা মোটেও টেকনিক্যাল ইন্ডিকেটর থেকে তৈরি হয় না, বরং একটি টুইট, একটি বক্তৃতা, একটি নীতি বিবৃতি সরাসরি দামকে ধাক্কা দেয়। টেকনিক্যাল অ্যানালাইসিস বলে "আপনি এখন কোথায় আছেন", কিন্তু খবরই প্রায়শই সেই মূল পরিবর্তনশীল যা বলে "আপনি এখানে কেন আছেন এবং পরবর্তীতে কোথায় যেতে পারেন"।
খবরের গুরুত্ব অনস্বীকার্য, কিন্তু সমস্যাটিও বাস্তব: কোনো মানুষ ২৪ ঘণ্টা চার্ট ও খবর দেখতে পারে না, তথ্যের স্রোতে সেটি খুঁজে বের করা আরও কঠিন যা সত্যিই দামের পরিবর্তন ঘটাবে। তাই একটি সরল চিন্তা মাথায় আসে – খবরগুলোকে সরাসরি K-লাইন চার্টে "আঁকা" যায় কি না, যাতে দাম এবং খবর একই ভিউতে একসাথে দেখা যায়? অন্তত "দেখা"র সমস্যাটি সমাধান করা যাক।
২. খবরের উৎস নির্বাচন: MCP এর মাধ্যমে জিনশি (Jin10) সংযোগ
খবর সংযুক্ত করার প্রথম ধাপ হলো এমন একটি উৎস খুঁজে বের করা যা যথেষ্ট দ্রুত আপডেট হয় এবং যার গঠন তুলনামূলকভাবে মানসম্মত। এখানে আমরা জিনশি ডেটা (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-লাইন একই চার্টে রাখা।
আমরা চার্টে একটি দ্বিতীয় সিরিজ যোগ করেছি, যার টাইপ 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()
প্রতি রিফ্রেশে, প্রথমে ক্রমবর্ধমানভাবে 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: কীওয়ার্ড অনুসারে খবরের flag, 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-লাইন বারের সাথে সংযুক্ত করা হয়, এবং বিষয়বস্তুর স্তরে কোনো ব্যাখ্যা করা হয় না। একটি খবর ভালো না খারাপ, এটি কি বাজারের গতিপথ পরিবর্তন করবে?—এটি সম্পূর্ণরূপে মানুষের বিচারের উপর নির্ভর করে।
দ্বিতীয়ত, কীওয়ার্ড ফিল্টারিং নিজেই একটি মোটামুটি সরল পদ্ধতি। একটি কীওয়ার্ড মিলে যাওয়ার অর্থ এই নয় যে খবরটি সত্যিই গুরুত্বপূর্ণ, এবং মিল না পাওয়ার অর্থ এই নয় যে এটি গুরুত্বহীন, এর মধ্যে ব্যক্তিগত জ্ঞান, অভিজ্ঞতা এবং এমনকি দিনের অবস্থা একটি বড় ওজন রাখতে পারে—একই খবর, বিভিন্ন মানুষ একই চার্ট দেখে, সম্পূর্ণ ভিন্ন সিদ্ধান্তে আসতে পারে।
তৃতীয়ত, পুরো প্রক্রিয়াটি এখনও "ম্যান-ইন-দ্যা-লুপ", প্রতিক্রিয়ার গতি মানুষের প্রতিক্রিয়ার গতি দ্বারা সীমাবদ্ধ, এবং অনেক সময় বাজারের খবরের প্রতিক্রিয়া মিনিট বা সেকেন্ডের স্তরে হয়।
যদি আপনারা এই দিকে আগ্রহী হন, আমরা ভবিষ্যতে একটি বড় ভাষার মডেল-ভিত্তিক অটোমেটেড সংস্করণ তৈরি করার চেষ্টা করব, যেখানে মডেল খবরের প্রাথমিক ব্যাখ্যা এবং গুরুত্ব নির্ধারণ করবে, এবং মানুষের সিদ্ধান্ত গ্রহণের সহায়তা বা এমনকি বিকল্প হিসেবে কাজ করবে। আগ্রহী হলে, অনুগ্রহ করে আমাদের অনুসরণ করতে থাকুন।
স্ট্র্যাটেজি সোর্স কোড: রিয়েল-টাইম নিউজ ফিউজ সিস্টেম
- 1