hyowons
fed3526b20
설정·스크립트·스킬·문서·큐레이션 메모리 추적. 시크릿(credentials/identity)·런타임 상태(state/logs/sessions/sqlite)· 백업(clobbered/bak)·dream 캐시는 .gitignore로 제외. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
334 lines
12 KiB
Python
334 lines
12 KiB
Python
#!/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 <code> [--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 텍스트 리스트. ▶/<br> 기준 분리, 태그 제거."""
|
|
if not html:
|
|
return []
|
|
t = html.replace('\r\n', '\n')
|
|
t = _re.sub(r'<br\s*/?>', '\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 <code> [--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()
|