#!/usr/bin/env python3 """WISEreport(FnGuide 계열) 컨센서스 조회 (조회 전용, 캐시 우선). FnGuide Snapshot(fnguide_client)이 못 주는 '시간축 컨센서스'를 보강한다: - 연도별 추정 재무 (IFRS연결, A 실적 + E 추정, 매출/영업이익/순이익/EPS/ROE + YoY) - 목표주가·추정치 리비전 추이 (최근 3개월 주간 — 상향/하향 모멘텀) - 어닝 서프라이즈 (시점별 컨센서스 vs 실적 괴리율) 데이터원: comp.wisereport.co.kr/company/ajax/{c1050001_data,cF5001}.aspx (JSON) ⚠️ 운영사가 FnGuide(WISEfn)로 fnguide_client와 동일 벤더. 같은 저작권·회색지대. → 종목별 파일 캐시(기본 12h) + 보유·관심 종목 on-demand 조회로만 사용. ⚠️ 데이터 스케일 주의: 현재 피드에서 forward 추정 절대값(매출·EPS)이 과거 실적 대비 크게 인플레이션돼 나오는 경우가 있다(현재가·목표가 동반). cutoff 가드는 두지 않고 소스값 그대로 반환한다(메모리 규칙). 방향성 신호(리비전 %·서프라이즈 %)는 스케일 무관. 호출부는 예외에 의존하지 않는다 — 실패 시 None 반환. CLI: python3 wisereport_client.py [--fresh] """ from __future__ import annotations import json import sys import time import urllib.request from pathlib import Path WORKSPACE = Path('/Users/snowoyh/.openclaw/agents/stock/workspace') CACHE_DIR = WORKSPACE / 'state' / 'wisereport_cache' CACHE_TTL_SEC = 12 * 3600 _BASE = 'https://comp.wisereport.co.kr/company/ajax' _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 # 서프라이즈/리비전 대상 지표 (acc_cd → 라벨) _SUP_ITEMS = {'121000': '매출액', '121500': '영업이익'} def _norm_code(code: str) -> str: s = ''.join(ch for ch in str(code) if ch.isdigit()) return s.zfill(6)[-6:] if s else '' def _num(s) -> float | None: if s is None: return None if isinstance(s, (int, float)): return float(s) t = str(s).strip().replace(',', '') if not t or t in ('-', 'N/A'): return None try: return float(t) except ValueError: return None def _today_kst() -> str: return time.strftime('%Y%m%d', time.localtime()) def _get_json(url: str, referer: str) -> dict | list | None: req = urllib.request.Request(url, headers={'User-Agent': _UA, 'Referer': referer}) try: with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp: raw = resp.read() except Exception: return None if not raw: return None try: return json.loads(raw.decode('utf-8', 'replace')) except Exception: return None def _parse_annual(code6: str, sdt: str, ref: str) -> list[dict]: """flag=2 → 연도별 추정 재무 (A 실적 + E 추정, IFRS연결).""" url = (f'{_BASE}/c1050001_data.aspx?cmp_cd={code6}&finGubun=MAIN&frq=0' f'&sDT={sdt}&chartType=svg&flag=2') d = _get_json(url, ref) rows = (d or {}).get('JsonData') if isinstance(d, dict) else None if not rows: return [] out = [] for r in rows: yymm = (r.get('YYMM') or '').strip() # "2025.12(A)" / "2026.12(E)" is_est = '(E)' in yymm period = yymm.replace('(A)', '').replace('(E)', '').replace('.', '/') out.append({ 'period': period, 'is_estimate': is_est, 'sales': _num(r.get('SALES')), 'sales_yoy': _num(r.get('YOY')), 'op': _num(r.get('OP')), 'np': _num(r.get('NP')), 'eps': _num(r.get('EPS')), 'bps': _num(r.get('BPS')), 'per': _num(r.get('PER')), 'pbr': _num(r.get('PBR')), 'roe': _num(r.get('ROE')), }) return out def _parse_revision(code6: str, sdt: str, yymm: str, ref: str) -> dict | None: """cF5001 → 목표주가·추정치(EPS) 주간 추이 + 변화율 요약.""" url = (f'{_BASE}/cF5001.aspx?cmp_cd={code6}&dt={sdt}&yymm={yymm}&frq=0' f'&acc_cd=121000&fingubun=MAIN&chartType=svg') d = _get_json(url, ref) if not isinstance(d, dict) or 'chart1' not in d: return None try: c = json.loads(d['chart1']) except Exception: return None dates = c.get('categories') or [] tp = [x for x in (c.get('target_price') or []) if isinstance(x, (int, float))] est = [x for x in (c.get('select_item') or []) if isinstance(x, (int, float))] def _chg(seq): if len(seq) >= 2 and seq[0]: return round((seq[-1] / seq[0] - 1) * 100, 1) return None if not tp and not est: return None return { 'item_name': c.get('select_item_name'), 'unit': c.get('select_item_unit'), 'dates': dates, 'target_price': c.get('target_price') or [], 'est_item': c.get('select_item') or [], 'target_first': tp[0] if tp else None, 'target_last': tp[-1] if tp else None, 'target_change_pct': _chg(tp), 'est_first': est[0] if est else None, 'est_last': est[-1] if est else None, 'est_change_pct': _chg(est), 'n_points': len(dates), } def _parse_surprise(code6: str, sdt: str, ref: str) -> dict | None: """flag=5 → 매출·영업이익 시점별 컨센서스 vs 실적 + 최신연도 서프라이즈%.""" items: dict[str, dict] = {} years = None for acc_cd, label in _SUP_ITEMS.items(): url = (f'{_BASE}/c1050001_data.aspx?cmp_cd={code6}&finGubun=MAIN&frq=0' f'&sDT={sdt}&chartType=svg&flag=5&acc_cd={acc_cd}') d = _get_json(url, ref) td = (d or {}).get('tableData') if isinstance(d, dict) else None rows = (td or {}).get('tableData') if isinstance(td, dict) else None if not rows: continue if years is None: hdr = (td.get('tableHeaderData') or [{}])[0] years = {k: hdr.get(k) for k in ('CNS_FY_2', 'CNS_FY_1', 'CNS_FY0', 'CNS_FY1')} # 최신 완료연도(FY0) 실적 + 발표직전(E) 서프라이즈% actual_fy0 = pre_surprise = pre_est = None for row in rows: qtr = (row.get('QTR') or '').strip() if qtr == '연간실적(A)': actual_fy0 = _num(row.get('FY0')) elif qtr == '발표직전(E)': pre_est = _num(row.get('FY0')) pre_surprise = _num(row.get('FY0_S')) # 괴리율 % items[label] = { 'fy0_year': (years or {}).get('CNS_FY0'), 'fy0_actual': actual_fy0, 'fy0_consensus': pre_est, 'fy0_surprise_pct': pre_surprise, } # 실데이터(연도 라벨 + 실적/서프라이즈) 하나도 없으면 None (ETF·없는종목). has_real = bool(years and any((years or {}).values())) and any( v.get('fy0_actual') is not None or v.get('fy0_surprise_pct') is not None for v in items.values() ) if not has_real: return None return {'years': years, 'items': items} def _build(code6: str) -> dict | None: sdt = _today_kst() ref = f'https://comp.wisereport.co.kr/company/c1050001.aspx?cmp_cd={code6}' annual = _parse_annual(code6, sdt, ref) # cF5001 yymm: 첫 추정연도(E) 기준, 없으면 최신연도 yymm = '000000' est_years = [a['period'] for a in annual if a.get('is_estimate')] if est_years: yymm = est_years[0].replace('/', '') # "2026/12" → "202612" elif annual: yymm = annual[-1]['period'].replace('/', '') revision = _parse_revision(code6, sdt, yymm, ref) surprise = _parse_surprise(code6, sdt, ref) if not annual and not revision and not surprise: return None return { 'code': code6, 'fetched_at': time.strftime('%Y-%m-%dT%H:%M:%S+09:00', time.localtime()), 'annual': annual, # IFRS연결, A+E 'revision': revision, # 목표주가·추정EPS 추이 'surprise': surprise, # 매출·영업이익 서프라이즈 } def _cache_path(code6: str) -> Path: return CACHE_DIR / f'{code6}.json' def get_consensus(code: str, max_age_sec: int = CACHE_TTL_SEC, force: bool = False) -> dict | None: """종목 컨센서스 dict 반환 (캐시 우선). 조회 불가 시 None — raise 안 함.""" 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 import re as _re def _clean_summary(html: str | None, max_bullets: int = 3) -> list[str]: """COMMENT2(HTML) → bullet 텍스트 리스트. ▶/
기준 분리, 태그 제거.""" if not html: return [] t = html.replace('\r\n', '\n') t = _re.sub(r'', '\n', t, flags=_re.I) t = _re.sub(r'<[^>]+>', '', t) # 태그 제거 t = t.replace('▶', '\n') bullets = [b.strip(' \t·-') for b in _re.split(r'\n+', t)] bullets = [b for b in bullets if b] return bullets[:max_bullets] def _reports_cache_path(code6: str) -> Path: return CACHE_DIR / f'{code6}_reports.json' def get_reports(code: str, limit: int = 8, max_age_sec: int = 6 * 3600, force: bool = False) -> list[dict] | None: """최근 증권사 분석리포트 목록. 조회 불가·없음 시 None — raise 안 함. 각 항목: {date, broker, broker_full, title, target, recomm, target_action, recomm_action, analyst, summary[]} 리포트는 수시 갱신 → 캐시 6h. PDF 원문은 게이팅 가능성으로 제외(메타+요약만). """ code6 = _norm_code(code) if not code6: return None cp = _reports_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 ref = f'https://comp.wisereport.co.kr/company/c1080001.aspx?cmp_cd={code6}' d = _get_json(f'{_BASE}/c1080001_data.aspx?cmp_cd={code6}', ref) rows = (d or {}).get('lists') if isinstance(d, dict) else None if not rows: return None out = [] for r in rows[:limit]: title = (r.get('PRE_TITLE') or '') + (r.get('RPT_TITLE') or '') out.append({ 'date': (r.get('ANL_DT') or '').strip(), 'broker': (r.get('BRK_NM_SHORT_KOR') or r.get('BRK_NM_KOR') or '').strip(), 'broker_full': (r.get('BRK_NM_KOR') or '').strip(), 'title': title.strip(), 'target': (r.get('TARGET_PRC') or '').strip() or None, 'recomm': (r.get('RECOMM') or '').strip() or None, 'target_action': (r.get('PRC_ACTION_TYP_NM') or '').strip() or None, 'recomm_action': (r.get('RECOMM_ACTION_TYP_NM') or '').strip() or None, 'analyst': (r.get('ANL_NM_KOR') or '').strip() or None, 'summary': _clean_summary(r.get('COMMENT2')), }) if not out: return None try: CACHE_DIR.mkdir(parents=True, exist_ok=True) tmp = cp.with_suffix('.json.tmp') tmp.write_text(json.dumps(out, ensure_ascii=False)) tmp.replace(cp) except Exception: pass return out def _cli() -> None: args = [a for a in sys.argv[1:] if not a.startswith('--')] if not args: print('usage: python3 wisereport_client.py [--fresh] [--reports]') sys.exit(1) if '--reports' in sys.argv: rs = get_reports(args[0], force=('--fresh' in sys.argv)) print(json.dumps(rs, ensure_ascii=False, indent=2) if rs else f'{args[0]}: 리포트 없음') sys.exit(0) d = get_consensus(args[0], force=('--fresh' in sys.argv)) 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()