From 3eab710dce87da08926c9a150b878da03e21b022 Mon Sep 17 00:00:00 2001 From: kyu Date: Wed, 3 Jun 2026 22:45:15 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=EA=B0=9C=EC=9D=B8=20=EC=88=98=EA=B8=89?= =?UTF-8?q?=20=EC=BB=A8=ED=8A=B8=EB=9E=98=EB=A6=AC=EC=96=B8=20=EC=A0=90?= =?UTF-8?q?=EC=88=98=20=EB=B0=98=EC=98=81=20(=EA=B8=B0=EA=B4=80+=EA=B0=9C?= =?UTF-8?q?=EC=9D=B8=20=EC=88=98=EA=B8=89=20=EC=8B=9C=EA=B7=B8=EB=84=90=20?= =?UTF-8?q?=C2=B115)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit inst_daily_flow.individual_net 5일 합산 추가. calc_inst_flow_signal에 개인 컨트래리언 로직: 개인 순매수=감점(과열), 개인이탈+큰손매집=가점, 개인매수+큰손매도=분산 감점. Co-Authored-By: Claude Opus 4.8 (1M context) --- score-engine/main.py | 28 ++++++++++++++++++++-------- 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/score-engine/main.py b/score-engine/main.py index 50086ca..d67baa4 100644 --- a/score-engine/main.py +++ b/score-engine/main.py @@ -2006,6 +2006,7 @@ async def _load_inst_flow_map(conn, as_of: date | None = None) -> dict: SELECT stock_code, SUM(inst_net) AS inst5d, SUM(foreign_net) AS for5d, + SUM(individual_net) AS ind5d, AVG(close_price)::float AS avg_price FROM inst_daily_flow WHERE trade_date >= CURRENT_DATE - 7 @@ -2016,6 +2017,7 @@ async def _load_inst_flow_map(conn, as_of: date | None = None) -> dict: SELECT stock_code, SUM(inst_net) AS inst5d, SUM(foreign_net) AS for5d, + SUM(individual_net) AS ind5d, AVG(close_price)::float AS avg_price FROM inst_daily_flow WHERE trade_date BETWEEN $1::date - 7 AND $1::date @@ -2025,25 +2027,35 @@ async def _load_inst_flow_map(conn, as_of: date | None = None) -> dict: def calc_inst_flow_signal(flow: dict) -> tuple[float, str]: - """기관 5일 순매수가 평균거래량 대비 의미 있으면 + 가산. 외국인과 같은 방향이면 추가 가중.""" + """기관+개인 수급 시그널 (±15). 기관 순매수 가산(외국인 동행 시 가중), + 개인은 컨트래리언: 개인 과열매수=감점, 개인이탈+큰손매집=가점.""" if not flow: return 0.0, "" inst5 = int(flow.get("inst5d") or 0) for5 = int(flow.get("for5d") or 0) - if inst5 == 0 and for5 == 0: + ind5 = int(flow.get("ind5d") or 0) + if inst5 == 0 and for5 == 0 and ind5 == 0: return 0.0, "" - # 기관 시그널: tanh 스케일 (포화) import math + # 기관 시그널: tanh 스케일 (포화) + 외국인 동행 가중 inst_score = math.tanh(inst5 / 5_000_000) * 6.0 - # 외국인 같은 방향이면 +4, 반대면 -2 if inst5 * for5 > 0: inst_score += 4.0 if abs(for5) > 1_000_000 else 2.0 elif inst5 * for5 < 0: inst_score -= 2.0 - sig = max(-10.0, min(10.0, inst_score)) - direction = "매수" if inst5 > 0 else "매도" - reason = f"기관 5d {direction}{abs(inst5)//10000:,}만주" - return sig, reason + # 개인 컨트래리언: 순매수→감점(과열), 순매도→가점 (소폭) + ind_score = -math.tanh(ind5 / 5_000_000) * 3.0 + smart = inst5 + for5 + if ind5 > 0 and smart < 0: ind_score -= 2.0 # 개인이 받고 큰손 던짐 = 분산 + elif ind5 < 0 and smart > 0: ind_score += 2.0 # 큰손 매집·개인 이탈 = 매집 + ind_score = max(-5.0, min(5.0, ind_score)) + sig = max(-15.0, min(15.0, inst_score + ind_score)) + parts = [] + if inst5 != 0: + parts.append(f"기관5d {'매수' if inst5>0 else '매도'}{abs(inst5)//10000:,}만주") + if abs(ind_score) >= 1.5: + parts.append(f"개인{'과열매수' if ind5>0 else '이탈'}") + return sig, " ".join(parts) def calc_valuation_percentile(per_history: list, cur_per: float) -> tuple[float, str]: