openclaw/agents/stock/workspace/scripts/fnguide_client.py

#!/usr/bin/env python3
"""FnGuide 컴퍼니가이드 펀더멘털 조회 (조회 전용, 캐시 우선).

키움 REST에 없는 데이터를 보강한다:
  - 연간 재무 시계열 (매출액·영업이익·순이익·EPS·ROE·PER) → 매출액/EPS/영업이익 증가율 계산
  - 컨센서스 (목표주가·투자의견·추정EPS·추정PER·추정 참여기관수)

데이터원: https://comp.fnguide.com/SVO2/xml/Snapshot_all/{code}.xml (EUC-KR XML)
FnGuide HTML은 이 XML을 JS로 렌더하므로 XML을 직접 받아 파싱한다.

⚠️ FnGuide 콘텐츠는 저작권 대상. 무단 대량수집·DB구축 금지.
   → 종목별 파일 캐시(기본 12h) + 보유·관심 종목 on-demand 조회로만 사용한다.

호출부는 절대 예외에 의존하지 않는다 — 실패 시 None 반환(분석은 FnGuide 없이도 진행).

CLI (수동 확인용):
  python3 fnguide_client.py <code>          # 펀더멘털 출력
  python3 fnguide_client.py <code> --fresh  # 캐시 무시 재조회
"""

from __future__ import annotations

import json
import sys
import time
import urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path

_SCRIPT_DIR = Path(__file__).resolve().parent
WORKSPACE = Path('/Users/snowoyh/.openclaw/agents/stock/workspace')
CACHE_DIR = WORKSPACE / 'state' / 'fnguide_cache'

CACHE_TTL_SEC = 12 * 3600          # 펀더멘털은 분기 단위 갱신 → 12h 캐시면 충분
SNAPSHOT_URL = 'https://comp.fnguide.com/SVO2/xml/Snapshot_all/{code6}.xml'
_UA = ('Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) '
       'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36')
_TIMEOUT = 8

# financial_highlight field(컬럼)명 prefix → 표준 키
_FIELD_MAP = {
    '매출액': 'sales',
    '영업이익(억원)': 'oper_profit',
    '당기순이익': 'net_profit',
    '부채비율': 'debt_ratio',
    '영업이익률': 'op_margin',
    '순이익률': 'net_margin',
    'ROA': 'roa',
    'ROE': 'roe',
    'EPS': 'eps',
    'BPS': 'bps',
    'PER': 'per',
    'PBR': 'pbr',
    '배당수익률': 'div_yield',
}


def _norm_code(code: str) -> str:
    """'A005930' / '005930' / 005930 → '005930' (6자리 숫자)."""
    s = ''.join(ch for ch in str(code) if ch.isdigit())
    return s.zfill(6)[-6:] if s else ''


def _num(s: str | None) -> float | None:
    if s is None:
        return None
    t = s.strip().replace(',', '')
    if not t or t in ('-', 'N/A', '_'):
        return None
    neg = t.startswith('(') and t.endswith(')')   # (1,234) = 음수 표기 대비
    if neg:
        t = t[1:-1]
    try:
        v = float(t)
        return -v if neg else v
    except ValueError:
        return None


def _txt(e) -> str:
    return (e.text or '').strip() if e is not None else ''


def _field_key(field_name: str) -> str | None:
    for prefix, key in _FIELD_MAP.items():
        if field_name.startswith(prefix):
            return key
    return None


def _pct_growth(cur: float | None, prev: float | None) -> float | None:
    """전년대비 증가율(%). 직전이 0/음수면 의미 없어 None."""
    if cur is None or prev is None or prev <= 0:
        return None
    return round((cur / prev - 1) * 100, 1)


def _parse_annual(root) -> list[dict]:
    """financial_highlight_annual → 실적 확정 연도만 시계열(오래된→최신).

    (E) 추정 연도는 별도 highlight value가 비어있어 제외(컨센서스는 _parse_consensus).
    """
    fha = root.find('.//financial_highlight_annual')
    if fha is None:
        return []
    fields = [_txt(f) for f in fha.findall('field')]
    keys = [_field_key(f) for f in fields]
    rows: list[dict] = []
    for rec in fha.findall('record'):
        period = _txt(rec.find('date'))
        if '(E)' in period:                 # 추정 연도 skip
            continue
        vals = [_txt(v) for v in rec.findall('value')]
        row: dict = {'period': period, 'fs_nm': _txt(rec.find('fs_nm'))}
        has_any = False
        for i, key in enumerate(keys):
            if key is None or i >= len(vals):
                continue
            n = _num(vals[i])
            row[key] = n
            if n is not None:
                has_any = True
        if has_any:
            rows.append(row)
    return rows


def _attach_growth(annual: list[dict]) -> None:
    """각 연도 행에 전년대비 증가율(%) 추가 (in-place)."""
    for i, row in enumerate(annual):
        prev = annual[i - 1] if i > 0 else None
        for base, out in (('sales', 'sales_growth'),
                          ('oper_profit', 'oper_profit_growth'),
                          ('eps', 'eps_growth'),
                          ('net_profit', 'net_profit_growth')):
            row[out] = _pct_growth(row.get(base), prev.get(base) if prev else None)


def _parse_consensus(root) -> dict | None:
    """<consensus> 블록 → 목표주가·투자의견·추정 EPS/PER·참여기관수."""
    c = root.find('.//consensus')
    if c is None:
        return None
    out = {
        'date': _txt(c.find('date')) or None,
        'target_price': _num(_txt(c.find('target_price'))),
        'opinion': _num(_txt(c.find('opinion'))),         # 1(매도)~5(매수) 스케일
        'eps': _num(_txt(c.find('eps'))),                 # 추정 EPS
        'per': _num(_txt(c.find('per'))),                 # 추정 PER
        'organ_count': _num(_txt(c.find('presume_organ_count'))),
    }
    # 전부 비면 의미 없음
    if not any(v is not None for k, v in out.items() if k != 'date'):
        return None
    return out


def _opinion_label(op: float | None) -> str | None:
    """투자의견 1~5 → 한글 라벨."""
    if op is None:
        return None
    if op >= 4.5:
        return '적극매수'
    if op >= 3.5:
        return '매수'
    if op >= 2.5:
        return '중립'
    if op >= 1.5:
        return '매도'
    return '적극매도'


def _fetch_xml(code6: str) -> str | None:
    url = SNAPSHOT_URL.format(code6=code6)
    req = urllib.request.Request(url, headers={
        'User-Agent': _UA,
        'Referer': 'https://comp.fnguide.com/',
    })
    try:
        with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
            raw = resp.read()
    except Exception:
        return None
    if not raw:
        return None
    return raw.decode('euc-kr', 'replace')


def _build(code6: str) -> dict | None:
    """XML fetch → 파싱 → 표준 dict. 실패/데이터 없으면 None."""
    xml_text = _fetch_xml(code6)
    if not xml_text:
        return None
    try:
        root = ET.fromstring(xml_text)
    except ET.ParseError:
        return None

    annual = _parse_annual(root)
    _attach_growth(annual)
    consensus = _parse_consensus(root)

    if not annual and not consensus:        # ETF·신규상장 등 펀더멘털 없음
        return None

    latest = annual[-1] if annual else {}
    growth = {
        'sales_yoy': latest.get('sales_growth'),
        'oper_profit_yoy': latest.get('oper_profit_growth'),
        'eps_yoy': latest.get('eps_growth'),
        'net_profit_yoy': latest.get('net_profit_growth'),
    } if latest else {}

    if consensus and consensus.get('opinion') is not None:
        consensus['opinion_label'] = _opinion_label(consensus['opinion'])

    return {
        'code': code6,
        'fetched_at': time.strftime('%Y-%m-%dT%H:%M:%S+09:00', time.localtime()),
        'annual': annual,            # 오래된→최신, 확정 실적만
        'latest_period': latest.get('period') if latest else None,
        'growth': growth,            # 최신 연도 YoY 요약
        'consensus': consensus,      # 목표주가·투자의견·추정치 (없으면 None)
    }


def _cache_path(code6: str) -> Path:
    return CACHE_DIR / f'{code6}.json'


def get_fundamentals(code: str, max_age_sec: int = CACHE_TTL_SEC,
                     force: bool = False) -> dict | None:
    """종목 펀더멘털 dict 반환 (캐시 우선). 조회 불가 시 None — 절대 raise 안 함.

    반환 스키마는 _build() 참조. ETF·없는 종목·네트워크 실패는 None.
    """
    code6 = _norm_code(code)
    if not code6:
        return None
    cp = _cache_path(code6)

    if not force and cp.exists():
        try:
            if time.time() - cp.stat().st_mtime < max_age_sec:
                return json.loads(cp.read_text())
        except Exception:
            pass  # 캐시 손상 → 재조회

    data = _build(code6)
    if data is None:
        return None
    try:
        CACHE_DIR.mkdir(parents=True, exist_ok=True)
        tmp = cp.with_suffix('.json.tmp')
        tmp.write_text(json.dumps(data, ensure_ascii=False))
        tmp.replace(cp)
    except Exception:
        pass  # 캐시 저장 실패해도 데이터는 반환
    return data


def _cli() -> None:
    args = [a for a in sys.argv[1:] if not a.startswith('--')]
    force = '--fresh' in sys.argv
    if not args:
        print('usage: python3 fnguide_client.py <code> [--fresh]')
        sys.exit(1)
    d = get_fundamentals(args[0], force=force)
    if d is None:
        print(f'{args[0]}: 펀더멘털 없음 (ETF/없는종목/조회실패)')
        sys.exit(0)
    print(json.dumps(d, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    _cli()