1. 시작: 트럼프가 직접 그린 K-선

뉴스의 시의성은 의심의 여지가 없습니다. 미국의 이란 공습 소식이 전해진 후, 원유 가격은 단시간에 급등했으며, 이 과정에서 트럼프와 이란의 발언 등 여러 요인이 지속적으로 얽히고 상호 강화되며 시세를 새로운 구간으로 계속 밀어 올렸습니다.

우리는 흔히 "트럼프가 직접 K-선을 그린다"고 농담을 합니다. 가격의 급격한 변동은 종종 기술적 지표에서 비롯된 것이 아니라, 하나의 트윗, 한 번의 연설, 한 번의 정책 발언이 직접 만들어낸 결과입니다. 기술적 분석은 "지금 어디에 있는지"를 알려주지만, 뉴스야말로 "왜 여기에 있으며, 앞으로 어디로 갈지"를 결정짓는 핵심 변수인 경우가 많습니다.

뉴스의 중요성은 부인할 수 없지만, 문제는 현실적입니다. 사람이 24시간 내내 차트와 뉴스를 주시할 수 없으며, 정보의 홍수 속에서 정말로 시세를 움직일 메시지를 즉시 포착하기는 더욱 어렵습니다. 그래서 아주 단순한 아이디어가 떠올랐습니다. 뉴스를 K-선 차트에 직접 "그려서" 가격과 메시지를 동일한 화면에서 동시에 보여줄 수 없을까? 최소한 "볼 수 있는" 문제부터 해결하자는 것입니다.

2. 뉴스 소스 선택: 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 신청이 필요합니다.

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

새로고침할 때마다 먼저 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-라인, 증분 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)

결과는 다음과 같습니다: 차트에 📰 마커가 나타날 때마다 마우스를 올리면 해당 뉴스 제목을 볼 수 있으며, 그 위치는 정확히 그 뉴스가 발생한 시간에 해당하는 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는 본질적으로 "정보 통합 + 수동 실행" 방식의 차트 모니터링 패널입니다. 가격, 뉴스, 포지션, 계좌 상태를 최대한 같은 화면에 배치하여 의사 결정은 전적으로 사람에게 맡기되, 판단 시 빠뜨리는 정보를 최소화하도록 돕습니다.

五、한계점: 사람이 여전히 가장 큰 변수

이 버전의 한계는 상당히 명확하며, 이를 회피하려 하지 않습니다.

첫째, 뉴스와 가격의 대응 관계는 "거친(granularity) 수준"입니다. 단순히 타임스탬프에 따라 뉴스를 해당 캔들 바에 매달았을 뿐, 내용 수준의 해석은 하지 않습니다. 한 뉴스가 호재인지 악재인지, 시장 움직임을 유발할지 여부는 전적으로 사람이 판단해야 합니다.

둘째, 키워드 필터링 자체는 비교적 단순한 방식입니다. 키워드가 적중했다고 해서 뉴스가 반드시 중요하다는 뜻이 아니며, 적중하지 않았다고 해서 중요하지 않다는 뜻도 아닙니다. 그 사이에서 개인의 인지 능력, 경험, 심지어 당일의 컨디션이 큰 비중을 차지할 수 있습니다. 같은 뉴스라도 다른 사람이 같은 차트를 보며 전혀 다른 결론을 내릴 수 있습니다.

셋째, 전체 프로세스는 여전히 "사람이 루프에 포함된(human-in-the-loop)" 방식이므로 응답 속도가 사람의 반응 속도에 제한되며, 실제로 뉴스에 대한 시장 반응은 분 단위, 심지어 초 단위로 발생하는 경우가 많습니다.

이 방향에 관심이 있으시다면, 향후 대규모 언어 모델(LLM) 기반의 자동화 버전을 시도하여 모델이 뉴스의 초기 해석과 중요도 판단을 수행하도록 함으로써 사람의 의사 결정을 보조하거나 대체할 수 있도록 할 계획입니다. 관심이 있다면 계속 지켜봐 주시기 바랍니다.

전략 소스 코드: 실시간 뉴스 트리거 시스템