QuantEngineByItz/tools/build_macro_context_from_workbook_v1.py

"""GatherTradingData.xlsx에서 실제 매크로/이벤트/포지션 컨텍스트를 추출.

build_qualitative_sell_inputs_v1.py의 --context-json을 수동 작성하지 않고, 이미
GAS 하네스가 산출/수집해 둔 시트 값을 그대로 읽어 자동 조립한다(중복 수집 금지
원칙 — qualitative_sell_strategy_v1.yaml:data_sources 참조).

실측 확인된 시트/컬럼(2026-06-21):
  - macro 시트: Symbol='MRS_COMPUTED'.Close = market_risk_score(0~10, 하네스 산출).
    Symbol='^TNX'(US10Y_Yield).Ret20D = 20일 금리추세 proxy(국내 기준금리 시트 없음 —
    한국은행 금통위 일정은 event_calendar Type='BOK'로 별도 포착).
  - event_risk 시트: Date/DaysLeft/Event/Type/Impact(HIGH/MEDIUM/LOW)/Alert/AsOfDate.
  - event_calendar 시트: Date/Event/Type(EARNINGS/FOMC/BOK/...)/Impact/DaysLeft 등.
    Type='EARNINGS'에 종목명이 Event 텍스트에 포함된 행만 종목별 실적발표일로 매칭.
  - account_snapshot 시트: ticker/name/holding_quantity/parse_status='CAPTURE_READ_OK'.
"""
from __future__ import annotations

import argparse
import datetime as dt
import json
import sys
from pathlib import Path
from typing import Any

from openpyxl import load_workbook

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

RATE_RISING_THRESHOLD_PCT = 2.0
RATE_FALLING_THRESHOLD_PCT = -2.0


def _read_sheet_rows(xlsx_path: Path, sheet: str) -> tuple[tuple, list[dict[str, Any]]]:
    """헤더 행을 탐색한다. 일부 시트(macro/event_risk)는 1행에 'updated: ...' 배너
    셀 1개만 있고 실제 헤더는 2행 — 비어있거나 단일 셀뿐인 선행 행은 건너뛴다."""
    wb = load_workbook(xlsx_path, read_only=True, data_only=True)
    ws = wb[sheet]
    rows_iter = ws.iter_rows(min_row=1, values_only=True)
    header: tuple = ()
    for row in rows_iter:
        non_empty = [c for c in row if c is not None]
        if len(non_empty) >= 2:
            header = row
            break
    rows = [dict(zip(header, row)) for row in rows_iter if any(c is not None for c in row)]
    return header, rows


def read_macro_pressure_and_regime(xlsx_path: Path) -> dict[str, Any]:
    """MRS_COMPUTED.Close(0~10) -> macro_pressure(-1~+1, 위험도 높을수록 매도압력).

    ^TNX Ret20D(%) -> rate_trend(RISING/FLAT/FALLING) — 국내 기준금리 시트가 없어
    미국채 10년물 20일 변화율을 proxy로 사용한다(국내 금리는 미 국채와 강한 동행성).
    """
    _, rows = _read_sheet_rows(xlsx_path, "macro")
    by_symbol = {row.get("Symbol"): row for row in rows}

    mrs_row = by_symbol.get("MRS_COMPUTED")
    macro_pressure = None
    market_risk_score = None
    if mrs_row is not None and isinstance(mrs_row.get("Close"), (int, float)):
        market_risk_score = float(mrs_row["Close"])
        macro_pressure = max(-1.0, min(1.0, (market_risk_score / 10.0) * 2.0 - 1.0))

    tnx_row = by_symbol.get("^TNX")
    rate_trend = None
    rate_ret20d_pct = None
    if tnx_row is not None and tnx_row.get("Ret20D") not in (None, ""):
        try:
            rate_ret20d_pct = float(tnx_row["Ret20D"])
        except (TypeError, ValueError):
            rate_ret20d_pct = None
        if rate_ret20d_pct is not None:
            if rate_ret20d_pct >= RATE_RISING_THRESHOLD_PCT:
                rate_trend = "RISING"
            elif rate_ret20d_pct <= RATE_FALLING_THRESHOLD_PCT:
                rate_trend = "FALLING"
            else:
                rate_trend = "FLAT"

    regime_row = by_symbol.get("REGIME_PRELIM")
    regime_prelim = regime_row.get("Close") if regime_row else None

    return {
        "macro_pressure": macro_pressure,
        "market_risk_score": market_risk_score,
        "rate_trend": rate_trend,
        "rate_ret20d_pct": rate_ret20d_pct,
        "regime_prelim": regime_prelim,
        "macro_pressure_source": "GatherTradingData.xlsx:macro",
    }


def read_next_macro_event(xlsx_path: Path, today: dt.date | None = None) -> dict[str, Any]:
    """event_risk 시트에서 오늘 이후 가장 가까운 HIGH 임팩트 이벤트일."""
    today = today or dt.date.today()
    _, rows = _read_sheet_rows(xlsx_path, "event_risk")
    candidates = []
    for row in rows:
        event_date = row.get("Date")
        if not isinstance(event_date, dt.datetime):
            continue
        event_date = event_date.date()
        if event_date < today or row.get("Impact") not in {"HIGH"}:
            continue
        candidates.append((event_date, row.get("Event"), row.get("Impact")))
    if not candidates:
        return {"next_macro_event_date": None, "macro_event_impact": None}
    candidates.sort(key=lambda item: item[0])
    event_date, event_name, impact = candidates[0]
    return {
        "next_macro_event_date": event_date.isoformat(),
        "macro_event_impact": impact,
        "macro_event_name": event_name,
        "macro_event_source": "GatherTradingData.xlsx:event_risk",
    }


def read_next_earnings_date(xlsx_path: Path, company_name: str, today: dt.date | None = None) -> dict[str, Any]:
    """event_calendar에서 Type='EARNINGS'이며 Event 텍스트에 종목명이 포함된 가장 빠른 미래 일정."""
    today = today or dt.date.today()
    _, rows = _read_sheet_rows(xlsx_path, "event_calendar")
    candidates = []
    name = (company_name or "").strip()
    if not name:
        return {"next_earnings_date": None, "earnings_event_impact": None}
    for row in rows:
        if row.get("Type") != "EARNINGS":
            continue
        event_text = str(row.get("Event") or "")
        if name not in event_text:
            continue
        event_date = row.get("Date")
        if isinstance(event_date, dt.datetime):
            event_date = event_date.date()
        elif isinstance(event_date, str):
            try:
                event_date = dt.date.fromisoformat(event_date)
            except ValueError:
                continue
        else:
            continue
        if event_date < today:
            continue
        candidates.append((event_date, row.get("Impact")))
    if not candidates:
        return {"next_earnings_date": None, "earnings_event_impact": None}
    candidates.sort(key=lambda item: item[0])
    event_date, impact = candidates[0]
    return {
        "next_earnings_date": event_date.isoformat(),
        "earnings_event_impact": impact,
        "earnings_source": "GatherTradingData.xlsx:event_calendar",
    }


def read_positions(xlsx_path: Path) -> list[dict[str, Any]]:
    """account_snapshot에서 실제 보유 종목 목록(CAPTURE_READ_OK, 보유수량>0)."""
    _, rows = _read_sheet_rows(xlsx_path, "account_snapshot")
    positions: dict[str, dict[str, Any]] = {}
    for row in rows:
        if row.get("parse_status") != "CAPTURE_READ_OK":
            continue
        ticker_raw = row.get("ticker")
        qty = row.get("holding_quantity") or 0
        if ticker_raw is None or not isinstance(qty, (int, float)) or qty <= 0:
            continue
        ticker = str(ticker_raw)
        ticker = ticker.zfill(6) if ticker.isdigit() else ticker
        entry = positions.setdefault(ticker, {"ticker": ticker, "name": row.get("name"), "holding_quantity": 0.0})
        entry["holding_quantity"] += float(qty)  # 소수주 분리 행 합산
    return list(positions.values())


def build_context_for_ticker(xlsx_path: Path, ticker: str, company_name: str) -> dict[str, Any]:
    today = dt.date.today()
    ctx: dict[str, Any] = {}
    ctx.update(read_macro_pressure_and_regime(xlsx_path))
    ctx.update(read_next_macro_event(xlsx_path, today))
    ctx.update(read_next_earnings_date(xlsx_path, company_name, today))
    return ctx


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--xlsx", type=Path, default=ROOT / "GatherTradingData.xlsx")
    ap.add_argument("--ticker", default=None)
    ap.add_argument("--name", default=None, help="실적발표 일정 매칭용 종목명(한글)")
    ap.add_argument("--list-positions", action="store_true")
    args = ap.parse_args()

    if args.list_positions:
        print(json.dumps(read_positions(args.xlsx), ensure_ascii=False, indent=2))
        return 0

    result = build_context_for_ticker(args.xlsx, args.ticker or "", args.name or "")
    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())