@@ -743,8 +743,9 @@ def calc_foreign_score(foreign_data: list) -> tuple[float, str]:
return max ( - 100.0 , min ( 100.0 , score ) ) , reason
def calc_short_score ( short_data : list ) - > tuple [ float , str ] :
""" 공매도 점수 (-100~+100), 공매도 많을수록 패널티 """
def calc_short_score ( short_data : list , rsi : float | None = None ) - > tuple [ float , str ] :
""" 공매도 점수 (-100~+100). 기본은 공매도 많을수록 패널티지만,
공매도 과다 + 과매도 ( RSI ≤ 35 ) 면 숏커버링 반등 ( 스퀴즈 ) 기대로 가산 — 양방향 . """
if not short_data :
return 0.0 , " "
score = 0.0
@@ -767,6 +768,12 @@ def calc_short_score(short_data: list) -> tuple[float, str]:
elif bal_chg_pct > 5 : score - = 10
elif bal_chg_pct < - 20 : score + = 15
elif bal_chg_pct < - 5 : score + = 7
# 숏스퀴즈 기대: 공매도 과다(거래비중≥5%) + 과매도(RSI≤35) → 숏커버링 반등 가능
# → 단방향 패널티를 상쇄/가산 (공매도 많을수록·과매도 깊을수록 가산 ↑, 최대 +35)
if rsi is not None and weight > = 5.0 and rsi < = 35.0 :
squeeze = min ( 35.0 , ( weight - 5.0 ) * 2.0 + ( 35.0 - rsi ) * 0.8 )
score + = squeeze
reason = ( reason or " " ) + f " 숏스퀴즈기대(RSI { rsi : .0f } ) "
return max ( - 100.0 , min ( 100.0 , score ) ) , reason
@@ -962,9 +969,14 @@ def calc_piotroski_score(curr: dict, prev: dict) -> tuple[int, float, str]:
# ── 알트만 Z-Score (단순화 — 운전자본·이익잉여금 데이터 부재) ─────
def calc_altman_z ( fin : dict , market_cap : int ) - > tuple [ float , str , str ] :
"""
Z ' ' 비제조업 모델 일부 변형 ( 가용 변수만 사용 )
Z ' ' 비제조업 모델의 2 항 변형 ( 운전자본 · 이익잉여금 데이터 부재로 X1 · X2 항 생략 ) .
Z_simple = 6.72 * ( EBIT / 총자산 ) + 1.05 * ( 시총 / 총부채 )
> 2.6 안전 / 1.1 ~ 2.6 회색 / < 1.1 부도위험
※ 임계값 재보정 : 원본 Z ' ' 의 안전선 2.6 은 4 항 ( X1 · X2 포함 ) 기준값이라
2 항만 쓰는 이 변형에 그대로 적용하면 Z가 체계적으로 낮게 나와 부도 신호가 과발생함
( 실측 : 전종목 중앙값 Z ≈ 2.0 , p10 ≈ 0.5 , p75 ≈ 5.1 인데 < 1.1 적용 시 ~ 31 % 가 ' 부도위험 ' ) .
→ 2 항 변형의 실측 분포에 맞춰 부도선 0.7 ( 하위 ~ 15 % ) , 안전선 3.0 ( 상위 ~ 40 % ) 로 보정 .
> 3.0 안전 / 0.7 ~ 3.0 회색 / < 0.7 부도위험
returns : ( z_score , signal ' 매수 ' | ' 매도 ' | ' 관망 ' , reason )
"""
op_pf = fin . get ( " operating_profit " , 0 ) or 0
@@ -975,9 +987,9 @@ def calc_altman_z(fin: dict, market_cap: int) -> tuple[float, str, str]:
a = op_pf / ta
b = ( market_cap / tl ) if tl > 0 else 1.0
z = 6.72 * a + 1.05 * b
if z > = 2.6 :
if z > = 3.0 :
return round ( z , 2 ) , " 매수 " , f " Altman Z { z : .1f } (안전) "
if z > = 1.1 :
if z > = 0.7 :
return round ( z , 2 ) , " 관망 " , " "
return round ( z , 2 ) , " 매도 " , f " Altman Z { z : .1f } (부도위험) "
@@ -1030,7 +1042,9 @@ async def calc_momentum(conn, stock_code: str, as_of: date | None = None) -> tup
if len ( closes ) < 200 :
return 0.0 , " 관망 " , " "
p_recent = closes [ 20 ] [ 1 ] # 1개월(거래일 ~21) 전
p_year = closes [ - 1 ] [ 1 ] # 약 12개월 전
# 12개월 전(거래일 252) 고정. closes[-1]을 쓰면 보유 데이터 길이(200~259)에 따라
# 룩백이 종목마다 달라져 모멘텀 비교가 불가능해짐 → 252봉으로 캡(부족 시 최장 사용).
p_year = closes [ min ( 251 , len ( closes ) - 1 ) ] [ 1 ]
if p_year < = 0 :
return 0.0 , " 관망 " , " "
mom = ( p_recent - p_year ) / p_year * 100
@@ -1999,6 +2013,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
@@ -2009,6 +2024,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
@@ -2018,25 +2034,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 ] :
@@ -2061,8 +2087,10 @@ def calc_valuation_percentile(per_history: list, cur_per: float) -> tuple[float,
return 0.0 , " "
async def calculate_daily_scores ( as_of : date | None = None ) :
async def calculate_daily_scores ( as_of : date | None = None , notify : bool = False ) :
""" 일간 점수 계산. as_of=None이면 today (운영 모드), as_of=date이면 그 시점 기준 (백필 모드).
notify = True일 때만 텔레그램 일간리포트 / 신규강력매수 발신 ( 16 : 30 정기 1 회만 True ) .
그 외 호출 ( 통합 워크플로우 30 분마다 · 매시간 스코어 · 18 : 30 등 ) 은 silent로 점수만 갱신 .
백필 모드는 look - ahead bias 차단 :
- 사후 학습 캐시 ( reliability / source_credibility ) 미적용
- weight_config는 config_date < = as_of 필터
@@ -2319,12 +2347,13 @@ async def calculate_daily_scores(as_of: date | None = None):
except : pass
# 기술적 점수 (stock_technical 테이블 - Redis TTL/DB 불일치 방지)
technical_score = 0.0
technical_score = 0.0 ; rsi_val = None
try :
ta_row = await conn . fetchrow (
" SELECT tech_score FROM stock_technical WHERE stock_code=$1 " , stock )
" SELECT tech_score, rsi FROM stock_technical WHERE stock_code=$1 " , stock )
if ta_row :
technical_score = float ( ta_row [ " tech_score " ] or 0 )
rsi_val = float ( ta_row [ " rsi " ] ) if ta_row [ " rsi " ] is not None else None
except : pass
# 외국인 수급 점수 (Redis foreign:{code})
@@ -2343,7 +2372,7 @@ async def calculate_daily_scores(as_of: date | None = None):
s_raw = await redis_cl . get ( f " short: { stock } " )
if s_raw :
s_data = json . loads ( s_raw )
short_score , short_reason = calc_short_score ( s_data )
short_score , short_reason = calc_short_score ( s_data , rsi = rsi_val )
short_weight_val = s_data [ 0 ] . get ( " trade_weight " , 0 ) if s_data else 0
except : pass
@@ -2818,7 +2847,7 @@ async def calculate_daily_scores(as_of: date | None = None):
logger . info ( " scoring.done " , scored = scored , recommended = len ( rec_rows ) )
# 신규 강력매수 즉시 알림 (어제는 강력매수 아니었는데 오늘 새로 등장한 종목)
if not backfill_mode and strong_buy :
if notify and not backfill_mode and strong_buy :
try :
async with pg_pool . acquire ( ) as nc :
prev_codes = { r [ " stock_code " ] for r in await nc . fetch ( """
@@ -2853,8 +2882,8 @@ async def calculate_daily_scores(as_of: date | None = None):
except Exception as e :
logger . warning ( " new_buy.err " , error = str ( e ) )
# 텔레그램 알림
if strong_buy or strong_sell :
# 텔레그램 알림 (notify=True 정기 1회만 — 매 호출 발신 방지)
if notify and ( strong_buy or strong_sell ) :
lines = [ f " <b>📊 Trading AI 일간 리포트 ( { today } )</b> \n " ]
if strong_buy :
lines . append ( " 🟢 <b>강력매수 추천 (버핏 가치필터 통과)</b> " )
@@ -2876,9 +2905,17 @@ async def calculate_daily_scores(as_of: date | None = None):
# ── 정기 브리핑 ───────────────────────────────────────────
_last_briefing_sent : datetime | None = None # 중복/폭주 차단 쓰로틀
async def send_briefing ( ) :
""" 정기 시황 브리핑 — 종목당 1카드(매매가/포지션/재무/근거 통합) """
""" 정기 시황 브리핑 — 종목당 1카드(매매가/포지션/재무/근거 통합).
30 분 내 재호출은 무시 — n8n 중복 / 재시도 폭주로 같은 브리핑이 여러 건 발송되는 것 방지 . """
global _last_briefing_sent
now = datetime . now ( )
if _last_briefing_sent and ( now - _last_briefing_sent ) . total_seconds ( ) < 1800 :
logger . info ( " briefing.throttled " , since = ( now - _last_briefing_sent ) . total_seconds ( ) )
return
_last_briefing_sent = now
ta_redis = aioredis . Redis (
host = REDIS_HOST , port = 6379 , password = REDIS_PASSWORD , db = 5 , decode_responses = True )
@@ -3018,6 +3055,17 @@ async def send_briefing():
app = FastAPI ( title = " 종목 점수 엔진 " )
app . add_middleware ( CORSMiddleware , allow_origins = [ " * " ] , allow_methods = [ " * " ] , allow_headers = [ " * " ] )
async def _daily_score_notify ( ) :
""" 16:30 정기 일간리포트 — 코루틴 함수로 등록해야 APScheduler가 await함
( lambda로 감싸면 coroutine never awaited 버그 ) . """
await calculate_daily_scores ( notify = True )
async def _learn_weights_job ( ) :
await learn_weights ( days = 90 , segment = " all " )
async def _learn_pricing_job ( ) :
await learn_pricing ( days = 90 , segment = " all " , target = " return_7d " , n_folds = 5 )
@app.on_event ( " startup " )
async def startup ( ) :
global pg_pool , redis_cl
@@ -3027,7 +3075,7 @@ async def startup():
redis_cl = aioredis . Redis (
host = REDIS_HOST , port = 6379 , password = REDIS_PASSWORD , db = 3 , decode_responses = True )
await init_db ( )
scheduler . add_job ( calculate_daily_scores , " cron " ,
scheduler . add_job ( _daily_score_notify , " cron " ,
day_of_week = " mon-fri " , hour = 16 , minute = 30 ,
id = " daily_score " , replace_existing = True )
# 텔레그램 정기 알림 하루 2회로 축소 (사용자 요청 2026-06-02):
@@ -3038,6 +3086,26 @@ async def startup():
scheduler . add_job ( cleanup_old_data , " cron " ,
hour = 4 , minute = 0 ,
id = " cleanup " , replace_existing = True )
# 데이터 무결성 모니터: 평일 10~17시 매시간 (RED면 텔레그램 경고, 3h 쓰로틀)
scheduler . add_job ( data_health_monitor_job , " cron " ,
day_of_week = " mon-fri " , hour = " 10-17 " , minute = 5 ,
id = " data_health " , replace_existing = True )
# 정확도 검증 리포트: 매주 일요일 10시 (방식이 실측 대비 맞는지)
scheduler . add_job ( accuracy_report_job , " cron " ,
day_of_week = " sun " , hour = 10 , minute = 0 ,
id = " accuracy_report " , replace_existing = True )
# 핫종목 검증팀: 평일 09:35 검증통과 핫종목 1회 보고 (dry-run)
scheduler . add_job ( hot_validate_report_job , " cron " ,
day_of_week = " mon-fri " , hour = 9 , minute = 35 ,
id = " hot_validate " , replace_existing = True )
# CIO 오늘의 결정안: 평일 09:20 생성+보고 (dry-run, 자동실행 OFF)
scheduler . add_job ( cio_decisions_job , " cron " ,
day_of_week = " mon-fri " , hour = 9 , minute = 20 ,
id = " cio_decisions " , replace_existing = True )
# 결정안 성과 채점: 매일 18:10 (성과가격 갱신 18:00 이후)
scheduler . add_job ( verify_decisions_job , " cron " ,
hour = 18 , minute = 10 ,
id = " verify_decisions " , replace_existing = True )
# 성과 추적: 매일 18시 가격 업데이트
scheduler . add_job ( update_performance_prices , " cron " ,
day_of_week = " mon-fri " , hour = 18 , minute = 0 ,
@@ -3050,11 +3118,10 @@ async def startup():
# 04:00 — 공식 가중치 학습 (90일 백테스트)
# 05:00 — 예상가 모델 학습 (선형회귀 + RF + XGBoost)
# 두 함수 모두 표본 부족 시 graceful (return early) — 데이터 누적되면 자동 활성화
scheduler . add_job ( lambda : learn_weights ( days = 90 , segment = " all " ) , " cron " ,
scheduler . add_job ( _learn_weights_job , " cron " ,
day_of_week = " sun " , hour = 4 , minute = 0 ,
id = " learn_weights " , replace_existing = True )
scheduler . add_job ( lambda : learn_pricing ( days = 90 , segment = " all " ,
target = " return_7d " , n_folds = 5 ) , " cron " ,
scheduler . add_job ( _learn_pricing_job , " cron " ,
day_of_week = " sun " , hour = 5 , minute = 0 ,
id = " learn_pricing " , replace_existing = True )
# AI 심층분석 — 비용 폭주로 자동 호출 임시 비활성화 (2026-05-29)
@@ -3144,13 +3211,425 @@ async def cleanup_old_data():
prices = deleted_prices , recs = deleted_recs ,
news = deleted_news , signals = deleted_signals )
# ── 데이터 무결성 모니터 ("데이터 제대로 쌓이는지" 상시 감시) ──────────
_last_health_alert : dict = { } # 동일 결함 반복경고 방지 (3시간 쓰로틀)
async def check_data_health ( ) - > dict :
""" 핵심 데이터 신선도·커버리지 점검 → 항목별 GREEN/YELLOW/RED. """
checks = [ ]
def add ( name , status , detail ) : checks . append ( { " name " : name , " status " : status , " detail " : detail } )
async with pg_pool . acquire ( ) as c :
n = await c . fetchval ( " SELECT COUNT(DISTINCT stock_code) FROM stock_prices WHERE collected_at::date=CURRENT_DATE " ) or 0
add ( " 시세(stock_prices) " , " GREEN " if n > = 2000 else " YELLOW " if n > = 500 else " RED " , f " 오늘 { n } 종목 " )
last_dt = await c . fetchval ( " SELECT MAX(dt) FROM stock_ohlcv WHERE stock_code<> ' KOSPI ' " )
ocnt = await c . fetchval ( " SELECT COUNT(DISTINCT stock_code) FROM stock_ohlcv WHERE dt=$1 " , last_dt ) if last_dt else 0
oage = ( date . today ( ) - last_dt ) . days if last_dt else 999
add ( " 일봉(stock_ohlcv) " , " GREEN " if ( ocnt > = 2000 and oage < = 4 ) else " YELLOW " if ocnt > = 1000 else " RED " , f " { last_dt } { ocnt } 종목 " )
n = await c . fetchval ( " SELECT COUNT(*) FROM stock_technical WHERE analyzed_at::date=CURRENT_DATE " ) or 0
add ( " 기술분석(TA) " , " GREEN " if n > = 1500 else " YELLOW " if n > = 300 else " RED " , f " 오늘 { n } 건 " )
n = await c . fetchval ( " SELECT COUNT(*) FROM stock_scores WHERE score_date=CURRENT_DATE " ) or 0
add ( " 점수(stock_scores) " , " GREEN " if n > = 1000 else " YELLOW " if n > = 100 else " RED " , f " 오늘 { n } 종목 " )
n = await c . fetchval ( " SELECT COUNT(*) FROM news_analysis WHERE analyzed_at > now()-interval ' 24 hours ' " ) or 0
add ( " 뉴스(24h) " , " GREEN " if n > = 100 else " YELLOW " if n > = 10 else " RED " , f " { n } 건 " )
kd = await c . fetchval ( " SELECT MAX(dt) FROM stock_ohlcv WHERE stock_code= ' KOSPI ' " )
kage = ( date . today ( ) - kd ) . days if kd else 999
add ( " KOSPI지수 일봉 " , " GREEN " if kage < = 4 else " RED " , f " { kd } " )
# OHLCV 이상치: 한국 일일 가격제한 ±30% 초과는 정의상 불량(스케일버그·권리락 미조정).
# 35% 마진으로 정상 상한가(±30%)는 제외. 데이터품질 경고라 YELLOW, 급증 시만 RED.
bad = await c . fetchval ( """
WITH px AS (
SELECT close_price , LAG ( close_price ) OVER ( PARTITION BY stock_code ORDER BY dt ) AS prev_close
FROM stock_ohlcv WHERE dt > CURRENT_DATE - 7 AND stock_code < > ' KOSPI '
)
SELECT COUNT ( * ) FROM px
WHERE close_price > 0 AND prev_close > 0 AND abs ( close_price : : float / prev_close - 1 ) > 0.35
""" ) or 0
add ( " OHLCV 이상치(±30 % 위반) " , " GREEN " if bad == 0 else " YELLOW " if bad < = 50 else " RED " , f " 최근7일 { bad } 행 " )
worst = " RED " if any ( x [ " status " ] == " RED " for x in checks ) else ( " YELLOW " if any ( x [ " status " ] == " YELLOW " for x in checks ) else " GREEN " )
return { " overall " : worst , " checks " : checks , " checked_at " : datetime . now ( ) . isoformat ( ) }
async def data_health_monitor_job ( ) :
""" 평일 장중~마감후 점검. RED 있으면 텔레그램 경고 1건 (3시간 쓰로틀). """
try :
h = await check_data_health ( )
except Exception as e :
logger . error ( " data_health.err " , error = str ( e ) ) ; return
reds = [ x for x in h [ " checks " ] if x [ " status " ] == " RED " ]
if not reds : return
key = " , " . join ( sorted ( x [ " name " ] for x in reds ) )
now = datetime . now ( )
last = _last_health_alert . get ( key )
if last and ( now - last ) . total_seconds ( ) < 10800 : return
_last_health_alert [ key ] = now
lines = [ " 🚨 <b>데이터 무결성 경고</b> " ] + [ f " ❌ { x [ ' name ' ] } : { x [ ' detail ' ] } " for x in reds ]
lines . append ( " \n 수집/분석 파이프라인 점검 필요 " )
await send_telegram ( " \n " . join ( lines ) )
logger . info ( " data_health.alert " , reds = len ( reds ) )
@app.get ( " /data-health " )
async def data_health_endpoint ( ) :
return await check_data_health ( )
# ── 정확도 검증 하베스트 ("방식이 맞는지" 실측 대비) ──────────────────
async def compute_accuracy ( days : int = 90 ) - > dict :
""" 추천 등급별 사후 정확도. recommendation_performance(실측 7d/30d 수익률·알파) 집계.
동전주 폭등 · 불량 가격데이터 이상치가 평균 ( mean ) 을 왜곡하므로 중앙값 ( median ) 으로 집계 .
매수계열 알파 > 0 & 매도계열 알파 < 0 이면 방식 유효 . """
async with pg_pool . acquire ( ) as conn :
grades = await conn . fetch ( """
SELECT recommendation rec , COUNT ( * ) n ,
percentile_cont ( 0.5 ) WITHIN GROUP ( ORDER BY return_7d ) ret7 ,
percentile_cont ( 0.5 ) WITHIN GROUP ( ORDER BY alpha_7d ) a7 ,
percentile_cont ( 0.5 ) WITHIN GROUP ( ORDER BY return_30d ) ret30 ,
percentile_cont ( 0.5 ) WITHIN GROUP ( ORDER BY alpha_30d ) a30 ,
AVG ( CASE WHEN return_7d > 0 THEN 1.0 ELSE 0 END ) up7
FROM recommendation_performance
WHERE return_7d IS NOT NULL AND rec_date > = CURRENT_DATE - ( $ 1 : : int )
GROUP BY recommendation
""" , days)
pooled = await conn . fetchrow ( """
SELECT percentile_cont ( 0.5 ) WITHIN GROUP ( ORDER BY alpha_7d )
FILTER ( WHERE recommendation IN ( ' 강력매수 ' , ' 매수관심 ' ) ) buy_a7 ,
percentile_cont ( 0.5 ) WITHIN GROUP ( ORDER BY alpha_7d )
FILTER ( WHERE recommendation IN ( ' 강력매도 ' , ' 매도관심 ' ) ) sell_a7 ,
percentile_cont ( 0.5 ) WITHIN GROUP ( ORDER BY alpha_30d )
FILTER ( WHERE recommendation IN ( ' 강력매수 ' , ' 매수관심 ' ) ) buy_a30 ,
percentile_cont ( 0.5 ) WITHIN GROUP ( ORDER BY alpha_30d )
FILTER ( WHERE recommendation IN ( ' 강력매도 ' , ' 매도관심 ' ) ) sell_a30 ,
percentile_cont ( 0.5 ) WITHIN GROUP ( ORDER BY alpha_30d )
FILTER ( WHERE recommendation = ' 강력매수 ' ) sb_a30 ,
percentile_cont ( 0.5 ) WITHIN GROUP ( ORDER BY alpha_30d )
FILTER ( WHERE recommendation = ' 강력매도 ' ) ss_a30
FROM recommendation_performance
WHERE return_7d IS NOT NULL AND rec_date > = CURRENT_DATE - ( $ 1 : : int )
""" , days)
order = { " 강력매수 " : 0 , " 매수관심 " : 1 , " 관망 " : 2 , " 매도관심 " : 3 , " 강력매도 " : 4 }
rows = sorted ( [ dict ( g ) for g in grades ] , key = lambda x : order . get ( x [ " rec " ] , 9 ) )
def rnd ( v ) : return round ( v , 2 ) if v is not None else None
buy_a , sell_a = rnd ( pooled [ " buy_a7 " ] ) , rnd ( pooled [ " sell_a7 " ] )
buy_a30 , sell_a30 = rnd ( pooled [ " buy_a30 " ] ) , rnd ( pooled [ " sell_a30 " ] )
sb30 , ss30 = rnd ( pooled [ " sb_a30 " ] ) , rnd ( pooled [ " ss_a30 " ] )
spread30 = round ( sb30 - ss30 , 2 ) if ( sb30 is not None and ss30 is not None ) else None
# 판정은 30일 기준(가치투자 시계열·이상치 robust median). 7일은 단기 노이즈라 참고용.
if buy_a30 is None or sell_a30 is None or spread30 is None :
verdict = " 30일 표본 부족 — 7일 참고 "
elif buy_a30 > 0 and sell_a30 < 0 and spread30 > = 5 :
verdict = " 양호 (30일 매수>0·매도<0·스프레드≥5 % p) "
elif spread30 > = 5 and sell_a30 < 0 :
verdict = " 부분유효 (강력매수 변별 양호, 매수계열 알파 음전) "
else :
verdict = " 교정필요 (30일 변별력 부족) "
return {
" days " : days ,
" agg " : " median " ,
" basis " : " 30d " ,
" grades " : [ { " rec " : r [ " rec " ] , " n " : r [ " n " ] ,
" ret7 " : round ( r [ " ret7 " ] or 0 , 2 ) , " alpha7 " : round ( r [ " a7 " ] or 0 , 2 ) ,
" ret30 " : round ( r [ " ret30 " ] , 2 ) if r [ " ret30 " ] is not None else None ,
" alpha30 " : round ( r [ " a30 " ] , 2 ) if r [ " a30 " ] is not None else None ,
" up7_pct " : round ( 100 * ( r [ " up7 " ] or 0 ) ) } for r in rows ] ,
" buy_alpha7 " : buy_a , " sell_alpha7 " : sell_a ,
" buy_alpha30 " : buy_a30 , " sell_alpha30 " : sell_a30 , " spread30 " : spread30 ,
" verdict " : verdict ,
}
@app.get ( " /accuracy " )
async def accuracy_endpoint ( days : int = Query ( default = 90 , ge = 7 , le = 365 ) ) :
return await compute_accuracy ( days )
async def accuracy_report_job ( ) :
""" 주간 정확도 리포트 — 방식이 실측 대비 맞는지 텔레그램 보고. """
try :
a = await compute_accuracy ( 90 )
except Exception as e :
logger . error ( " accuracy.err " , error = str ( e ) ) ; return
lines = [ " 📈 <b>추천 정확도 리포트 (최근90일·30일 알파·중앙값)</b> " ,
f " 판정: <b> { a [ ' verdict ' ] } </b> " ,
f " 매수계열 알파 { a [ ' buy_alpha30 ' ] } / 매도계열 알파 { a [ ' sell_alpha30 ' ] } / 강력매수−강력매도 스프레드 { a [ ' spread30 ' ] } %p \n " ]
for g in a [ " grades " ] :
lines . append ( f " { g [ ' rec ' ] } : n { g [ ' n ' ] } 30일수익 { g [ ' ret30 ' ] } % 알파 { g [ ' alpha30 ' ] } % (7일알파 { g [ ' alpha7 ' ] } %) " )
await send_telegram ( " \n " . join ( lines ) )
logger . info ( " accuracy.report.sent " , verdict = a [ " verdict " ] )
# ── 🔥 핫종목 검증팀 (키움 핫 × 가치/품질 노이즈필터, dry-run) ──────────
async def compute_hot_validate ( top : int = 30 ) - > dict :
""" 키움 거래량급증(ka10023) × 가치/품질 필터 → 핫한데 가치없는 노이즈
( ETF · 파생 · 작전 · 적자 · 분식 · 음수점수 ) 제거 → ' 검증통과 ' 만 추림 . """
hot = [ ]
try :
async with httpx . AsyncClient ( ) as cli :
r = await cli . get ( " http://kis-api:8585/volume-surge " , timeout = 15 )
hot = ( ( r . json ( ) or { } ) . get ( " data " ) or [ ] ) [ : top ]
except Exception as e :
logger . warning ( " hot_validate.fetch_err " , error = str ( e ) )
return { " error " : " 키움 핫종목 조회 실패 " , " detail " : str ( e ) }
codes = [ h [ " code " ] for h in hot if h . get ( " code " ) ]
if not codes :
return { " hot_count " : 0 , " passed " : [ ] , " noise " : [ ] , " watch " : [ ] }
async with pg_pool . acquire ( ) as conn :
rows = await conn . fetch ( """
SELECT h . code , c . corp_name , c . is_active ,
sc . total_score , sc . recommendation , sc . beneish_score ,
pr . market_cap , fin . operating_profit
FROM unnest ( $ 1 : : text [ ] ) AS h ( code )
LEFT JOIN dart_corps c ON c . stock_code = h . code
LEFT JOIN LATERAL ( SELECT total_score , recommendation , beneish_score
FROM stock_scores WHERE stock_code = h . code
ORDER BY score_date DESC LIMIT 1 ) sc ON true
LEFT JOIN LATERAL ( SELECT market_cap FROM stock_prices WHERE stock_code = h . code
ORDER BY collected_at DESC LIMIT 1 ) pr ON true
LEFT JOIN LATERAL ( SELECT operating_profit FROM dart_financials
WHERE stock_code = h . code AND reprt_code = ' 11011 '
ORDER BY bsns_year DESC LIMIT 1 ) fin ON true
""" , codes)
by = { r [ " code " ] : r for r in rows }
passed , noise , watch = [ ] , [ ] , [ ]
for h in hot :
code = h . get ( " code " ) ; r = by . get ( code )
it = { " code " : code , " name " : ( r [ " corp_name " ] if r and r [ " corp_name " ] else h . get ( " name " , " " ) ) ,
" change_pct " : h . get ( " change_pct " ) , " surge_rate " : h . get ( " surge_rate " ) ,
" score " : round ( r [ " total_score " ] , 1 ) if r and r [ " total_score " ] is not None else None ,
" reco " : r [ " recommendation " ] if r else None }
if not r or not r [ " is_active " ] :
it [ " verdict " ] , it [ " reason " ] = " 노이즈 " , " ETF/파생/비상장(가치판단 불가) " ; noise . append ( it ) ; continue
mc , op , sc , bn = r [ " market_cap " ] or 0 , r [ " operating_profit " ] , r [ " total_score " ] , r [ " beneish_score " ]
if mc and mc < 10_000_000_000 :
it [ " verdict " ] , it [ " reason " ] = " 노이즈 " , f " 초소형 시총 { mc / / 100000000 } 억(작전위험) " ; noise . append ( it ) ; continue
if op is not None and op < = 0 :
it [ " verdict " ] , it [ " reason " ] = " 노이즈 " , " 영업적자 " ; noise . append ( it ) ; continue
if bn is not None and bn > = 50 :
it [ " verdict " ] , it [ " reason " ] = " 노이즈 " , f " 분식의심(Beneish { bn : .0f } ) " ; noise . append ( it ) ; continue
if sc is None :
it [ " verdict " ] , it [ " reason " ] = " 관찰 " , " 점수 미산출 " ; watch . append ( it ) ; continue
if sc < 0 :
it [ " verdict " ] , it [ " reason " ] = " 노이즈 " , f " 투자부적합(점수 { sc : .0f } ) " ; noise . append ( it ) ; continue
if sc > = 40 and r [ " recommendation " ] in ( " 강력매수 " , " 매수관심 " ) :
it [ " verdict " ] , it [ " reason " ] = " 검증통과 " , f " 가치+모멘텀(점수 { sc : .0f } · { r [ ' recommendation ' ] } ) " ; passed . append ( it ) ; continue
it [ " verdict " ] , it [ " reason " ] = " 관찰 " , f " 중립(점수 { sc : .0f } ) " ; watch . append ( it )
return { " hot_count " : len ( hot ) , " passed_count " : len ( passed ) ,
" noise_count " : len ( noise ) , " watch_count " : len ( watch ) ,
" noise_filtered_pct " : round ( 100 * len ( noise ) / len ( hot ) ) if hot else 0 ,
" passed " : passed , " watch " : watch , " noise " : noise }
@app.get ( " /hot/validate " )
async def hot_validate_endpoint ( top : int = Query ( default = 30 , ge = 5 , le = 50 ) ) :
return await compute_hot_validate ( top )
async def hot_validate_report_job ( ) :
""" 평일 09:35 — 검증통과 핫종목만 텔레그램 보고 (dry-run, 1일 1회). """
try :
v = await compute_hot_validate ( 30 )
except Exception as e :
logger . error ( " hot_validate.err " , error = str ( e ) ) ; return
if v . get ( " error " ) or not v . get ( " passed " ) :
return
lines = [ f " 🔥 <b>검증통과 핫종목</b> (키움 핫 { v [ ' hot_count ' ] } 개 중 노이즈 { v [ ' noise_filtered_pct ' ] } % 제거) " ]
for p in v [ " passed " ] [ : 10 ] :
lines . append ( f " ✅ { p [ ' name ' ] } ( { p [ ' code ' ] } ) { p [ ' change_pct ' ] : +.1f } % — { p [ ' reason ' ] } " )
await send_telegram ( " \n " . join ( lines ) )
logger . info ( " hot_validate.report.sent " , passed = v [ " passed_count " ] )
# ── 🧭 CIO 종합 에이전트 (오늘의 결정안, dry-run) ──────────────────
def _conviction ( score : float , votes : int ) - > int :
if score > = 70 and votes > = 3 : return 5
if score > = 55 and votes > = 2 : return 4
if score > = 45 : return 3
if score > = 35 : return 2
return 1
async def run_cio_decisions ( top : int = 8 ) - > dict :
""" 점수+보팅+리스크+핫검증 종합 → 오늘의 매수/매도 결정안 (dry-run, 승인 전).
매수 = 추천 상위 후보 , 매도 = 보유종목 ( user_portfolio ) 등급악화 · 손절 . entry_price 기록 ( 사후추적 ) . """
today = date . today ( )
async with pg_pool . acquire ( ) as conn :
await conn . execute ( """
CREATE TABLE IF NOT EXISTS daily_decisions (
id SERIAL PRIMARY KEY ,
decision_date DATE NOT NULL ,
stock_code VARCHAR ( 10 ) NOT NULL ,
stock_name VARCHAR ( 100 ) DEFAULT ' ' ,
action VARCHAR ( 20 ) NOT NULL ,
conviction INTEGER DEFAULT 0 ,
size_pct FLOAT DEFAULT 0 ,
total_score FLOAT ,
thesis TEXT DEFAULT ' ' ,
risk_notes TEXT DEFAULT ' ' ,
status VARCHAR ( 20 ) DEFAULT ' proposed ' ,
created_at TIMESTAMP DEFAULT NOW ( ) ,
UNIQUE ( decision_date , stock_code )
)
""" )
for col in ( " entry_price BIGINT DEFAULT 0 " , " return_7d FLOAT " ,
" kospi_return_7d FLOAT " , " alpha_7d FLOAT " , " outcome VARCHAR(10) " ) :
await conn . execute ( f " ALTER TABLE daily_decisions ADD COLUMN IF NOT EXISTS { col } " )
regime_label , _ = await calc_market_regime ( conn )
hotv = await compute_hot_validate ( 30 )
hot_pass = { p [ " code " ] for p in ( hotv . get ( " passed " ) or [ ] ) } if not hotv . get ( " error " ) else set ( )
async def _cur_price ( code ) :
return await conn . fetchval (
" SELECT price FROM stock_prices WHERE stock_code=$1 ORDER BY collected_at DESC LIMIT 1 " , code )
saved = [ ]
# ── 매수 결정안 (추천 상위 후보) ──
cands = await conn . fetch ( """
SELECT s . stock_code , COALESCE ( d . corp_name , s . stock_code ) name ,
s . total_score , s . recommendation , s . buy_votes ,
s . position_size_pct , s . top_reasons , s . beneish_score , s . earnings_quality
FROM stock_scores s JOIN dart_corps d ON d . stock_code = s . stock_code
WHERE s . score_date = $ 1 AND d . is_active = true
AND s . recommendation IN ( ' 강력매수 ' , ' 매수관심 ' ) AND s . buy_votes > = 1
ORDER BY s . total_score DESC LIMIT $ 2
""" , today, top)
for c in cands :
conv = _conviction ( c [ " total_score " ] or 0 , c [ " buy_votes " ] or 0 )
if c [ " stock_code " ] in hot_pass : conv = min ( 5 , conv + 1 )
size = round ( c [ " position_size_pct " ] or ( conv * 2.0 ) , 1 )
if regime_label == " 약세 " : size = round ( size * 0.5 , 1 )
risk = [ ]
if regime_label == " 약세 " : risk . append ( " 시장 약세(사이즈↓) " )
if ( c [ " beneish_score " ] or 0 ) > = 50 : risk . append ( " 분식의심 " )
if c [ " earnings_quality " ] is not None and c [ " earnings_quality " ] < 0 : risk . append ( " 이익품질 낮음 " )
if c [ " stock_code " ] in hot_pass : risk . append ( " 핫종목 검증통과 " )
ep = await _cur_price ( c [ " stock_code " ] ) or 0
await conn . execute ( """
INSERT INTO daily_decisions
( decision_date , stock_code , stock_name , action , conviction , size_pct ,
total_score , thesis , risk_notes , entry_price , status )
VALUES ( $ 1 , $ 2 , $ 3 , ' 매수 ' , $ 4 , $ 5 , $ 6 , $ 7 , $ 8 , $ 9 , ' proposed ' )
ON CONFLICT ( decision_date , stock_code ) DO UPDATE SET
action = ' 매수 ' , conviction = $ 4 , size_pct = $ 5 , total_score = $ 6 ,
thesis = $ 7 , risk_notes = $ 8 , entry_price = $ 9
""" , today, c[ " stock_code " ], c[ " name " ], conv, size,
c [ " total_score " ] , ( c [ " top_reasons " ] or " " ) [ : 300 ] , " · " . join ( risk ) , int ( ep ) )
saved . append ( { " code " : c [ " stock_code " ] , " name " : c [ " name " ] , " action " : " 매수 " ,
" conviction " : conv , " size_pct " : size ,
" score " : round ( c [ " total_score " ] , 1 ) if c [ " total_score " ] is not None else None ,
" risk " : " · " . join ( risk ) } )
# ── 매도 결정안 (보유종목 등급악화/손절) ──
holds = await conn . fetch ( """
SELECT p . stock_code , p . stock_name , p . buy_price ,
s . total_score , s . recommendation , s . sell_votes , s . top_reasons
FROM user_portfolio p
LEFT JOIN stock_scores s ON s . stock_code = p . stock_code AND s . score_date = $ 1
WHERE p . active = true
""" , today)
for h in holds :
cp = await _cur_price ( h [ " stock_code " ] ) or 0
loss = ( ( cp - h [ " buy_price " ] ) / h [ " buy_price " ] * 100 ) if ( h [ " buy_price " ] and cp ) else 0
reco = h [ " recommendation " ] ; reasons = [ ] ; sell = False
if reco in ( " 강력매도 " , " 매도관심 " ) : sell = True ; reasons . append ( f " 등급 { reco } " )
if loss < = - 8 : sell = True ; reasons . append ( f " 손절( { loss : .0f } %) " )
if ( h [ " sell_votes " ] or 0 ) > = 3 : sell = True ; reasons . append ( f " 매도보팅 { h [ ' sell_votes ' ] } " )
if not sell : continue
conv = 5 if ( reco == " 강력매도 " or loss < = - 12 ) else 3
await conn . execute ( """
INSERT INTO daily_decisions
( decision_date , stock_code , stock_name , action , conviction , size_pct ,
total_score , thesis , risk_notes , entry_price , status )
VALUES ( $ 1 , $ 2 , $ 3 , ' 매도 ' , $ 4 , 0 , $ 5 , $ 6 , $ 7 , $ 8 , ' proposed ' )
ON CONFLICT ( decision_date , stock_code ) DO UPDATE SET
action = ' 매도 ' , conviction = $ 4 , total_score = $ 5 , thesis = $ 6 , risk_notes = $ 7 , entry_price = $ 8
""" , today, h[ " stock_code " ], h[ " stock_name " ], conv,
h [ " total_score " ] , ( h [ " top_reasons " ] or " " ) [ : 200 ] , " · " . join ( reasons ) , int ( cp ) )
saved . append ( { " code " : h [ " stock_code " ] , " name " : h [ " stock_name " ] , " action " : " 매도 " ,
" conviction " : conv , " size_pct " : 0 ,
" score " : round ( h [ " total_score " ] , 1 ) if h [ " total_score " ] is not None else None ,
" risk " : " · " . join ( reasons ) } )
return { " decision_date " : str ( today ) , " regime " : regime_label , " count " : len ( saved ) ,
" buy " : sum ( 1 for x in saved if x [ " action " ] == " 매수 " ) ,
" sell " : sum ( 1 for x in saved if x [ " action " ] == " 매도 " ) ,
" decisions " : saved }
@app.post ( " /decisions/generate " )
async def decisions_generate ( top : int = Query ( default = 8 , ge = 1 , le = 20 ) ) :
return await run_cio_decisions ( top )
@app.get ( " /decisions " )
async def decisions_get ( ) :
async with pg_pool . acquire ( ) as conn :
rows = await conn . fetch ( """
SELECT stock_code , stock_name , action , conviction , size_pct , total_score ,
thesis , risk_notes , status
FROM daily_decisions WHERE decision_date = CURRENT_DATE
ORDER BY conviction DESC , total_score DESC
""" )
return [ dict ( r ) for r in rows ]
async def verify_decisions_job ( ) :
""" 결정안 7일 성과 채점 (return/alpha/정답여부) — ' 회사 결정이 실제 맞았나 ' . 매일 18:10. """
scored = 0
async with pg_pool . acquire ( ) as conn :
rows = await conn . fetch ( """
SELECT id , stock_code , action , entry_price , decision_date
FROM daily_decisions
WHERE return_7d IS NULL AND entry_price > 0
AND decision_date < = CURRENT_DATE - 7 AND decision_date > = CURRENT_DATE - 60
""" )
for r in rows :
target = r [ " decision_date " ] + timedelta ( days = 7 )
if target > date . today ( ) : continue
price = await _close_near ( conn , r [ " stock_code " ] , target )
if price is None : continue
ret = ( price - r [ " entry_price " ] ) / r [ " entry_price " ] * 100
kret = await _kospi_return_between ( conn , r [ " decision_date " ] , target )
alpha = ( ret - kret ) if kret is not None else None
outcome = None
if alpha is not None :
outcome = ( " 정답 " if alpha > 0 else " 오답 " ) if r [ " action " ] == " 매수 " \
else ( " 정답 " if alpha < 0 else " 오답 " )
await conn . execute ( """ UPDATE daily_decisions
SET return_7d = $ 1 , kospi_return_7d = $ 2 , alpha_7d = $ 3 , outcome = $ 4 WHERE id = $ 5 """ ,
ret , kret , alpha , outcome , r [ " id " ] )
scored + = 1
logger . info ( " verify_decisions.done " , scored = scored )
@app.get ( " /decisions/accuracy " )
async def decisions_accuracy ( days : int = Query ( default = 60 , ge = 7 , le = 365 ) ) :
""" CIO 결정안의 실측 성과 — 자동실행 게이트. """
async with pg_pool . acquire ( ) as conn :
rows = await conn . fetch ( """
SELECT action , COUNT ( * ) n , AVG ( return_7d ) ret , AVG ( alpha_7d ) alpha ,
AVG ( CASE WHEN outcome = ' 정답 ' THEN 1.0 ELSE 0 END ) hit
FROM daily_decisions
WHERE return_7d IS NOT NULL AND decision_date > = CURRENT_DATE - ( $ 1 : : int )
GROUP BY action
""" , days)
return { " days " : days , " by_action " : [
{ " action " : r [ " action " ] , " n " : r [ " n " ] ,
" avg_return7 " : round ( r [ " ret " ] or 0 , 2 ) , " avg_alpha7 " : round ( r [ " alpha " ] or 0 , 2 ) ,
" hit_rate " : round ( 100 * ( r [ " hit " ] or 0 ) ) } for r in rows ] }
async def cio_decisions_job ( ) :
""" 평일 09:20 — CIO 오늘의 결정안 생성 + 텔레그램 보고 (dry-run, 자동실행 OFF). """
try :
d = await run_cio_decisions ( 8 )
except Exception as e :
logger . error ( " cio.err " , error = str ( e ) ) ; return
if not d . get ( " decisions " ) : return
lines = [ f " 🧭 <b>오늘의 결정안</b> ( { d [ ' decision_date ' ] } , 시장: { d [ ' regime ' ] } ) — dry-run " ]
for x in d [ " decisions " ] :
lines . append ( f " { ' ★ ' * x [ ' conviction ' ] } <b> { x [ ' name ' ] } </b>( { x [ ' code ' ] } ) { x [ ' action ' ] } "
f " 비중 { x [ ' size_pct ' ] } % (점수 { x [ ' score ' ] } ) "
+ ( f " \n ⚠️ { x [ ' risk ' ] } " if x [ ' risk ' ] else " " ) )
lines . append ( " \n (자동실행 OFF — 검증 단계. 정확도 양전환 시 실행 연결) " )
await send_telegram ( " \n " . join ( lines ) )
logger . info ( " cio.report.sent " , count = d [ " count " ] )
@app.get ( " /health " )
async def health ( ) :
return { " status " : " ok " }
@app.post ( " /score/calculate " )
async def manual_calc ( ) :
n = await calculate_daily_scores ( )
async def manual_calc ( notify : bool = Query ( default = False ) ) :
n = await calculate_daily_scores ( notify = notify )
return { " status " : " done " , " scored " : n }
@@ -4038,22 +4517,43 @@ def _eval_metrics(y_true, y_pred) -> dict:
return out
def _walk_forward_folds ( rows , n_folds : int ) :
def _walk_forward_folds ( rows , n_folds : int , embargo_days : int = 0 ) :
"""
시간순 정렬된 rows를 ( n_folds + 1 ) 블록으로 나누어 expanding window 폴드 생성 .
각 fold i ∈ [ 1. . n_folds ] : train = [ 0. . i * block ] , test = [ i * block . . ( i + 1 ) * block ] .
test 데이터는 학습에 절대 안 들어감 → leakage 없음 .
시간순 정렬된 rows를 날짜 ( score_date ) 경계로 ( n_folds + 1 ) 블록으로 나눈 expanding - window 폴드 .
각 fold i : test = 다음 날짜블록 , train = 그 이전 날짜들 . 단 ,
① 같은 score_date가 train / test에 쪼개지지 않도록 ' 날짜 ' 단위로 분할
( 인덱스 분할은 한 날짜의 종목들이 train / test로 갈라져 시장 단면 누수 발생 )
② train의 마지막 embargo_days 구간은 purge — train 라벨의 미래 수익 윈도 ( + N일 ) 가
test 피처 시점과 겹치는 패널 데이터 누수 ( López de Prado purge / embargo ) 차단 .
"""
n = len ( rows )
if n < ( n_folds + 1 ) * 3 :
return [ ]
dates = sorted ( { r [ " score_date " ] for r in rows } )
if len ( dates ) < n_folds + 1 :
# 고유 날짜가 적으면 날짜 분할 불가 → 인덱스 분할로 폴백(임바고는 그대로 적용)
block = n / / ( n_folds + 1 )
folds = [ ]
for i in range ( 1 , n_folds + 1 ) :
tr_end = i * block
te_end = min ( n , ( i + 1 ) * block )
tr = rows [ : tr_end ] ; te = rows [ tr_end : te_end ]
if len ( tr ) < 5 or len ( te ) < 3 : continue
te = rows [ i * block : min ( n , ( i + 1 ) * block ) ]
if not te :
continue
emb_cut = te [ 0 ] [ " score_date " ] - timedelta ( days = embargo_days )
tr = [ r for r in rows [ : i * block ] if r [ " score_date " ] < = emb_cut ]
if len ( tr ) < 5 or len ( te ) < 3 :
continue
folds . append ( ( tr , te ) )
return folds
dblock = len ( dates ) / / ( n_folds + 1 )
folds = [ ]
for i in range ( 1 , n_folds + 1 ) :
te_start_date = dates [ i * dblock ]
te_end_date = dates [ min ( len ( dates ) , ( i + 1 ) * dblock ) - 1 ]
emb_cut = te_start_date - timedelta ( days = embargo_days )
tr = [ r for r in rows if r [ " score_date " ] < = emb_cut ]
te = [ r for r in rows if te_start_date < = r [ " score_date " ] < = te_end_date ]
if len ( tr ) < 5 or len ( te ) < 3 :
continue
folds . append ( ( tr , te ) )
return folds
@@ -4103,33 +4603,44 @@ async def learn_pricing(days: int = Query(default=180, ge=14, le=730),
import numpy as np
from sklearn . linear_model import Ridge
from sklearn . ensemble import RandomForestRegressor
from sklearn . preprocessing import StandardScaler
from sklearn . pipeline import make_pipeline
except Exception as e :
return { * * out , " err " : f " sklearn import 실패: { e } " }
folds = _walk_forward_folds ( rows , n_folds )
# 라벨 수익 윈도 길이만큼 train↔test 사이 임바고 (7d→+11일, 30d→+38일 윈도 → +1 여유)
embargo_days = { " return_7d " : 12 , " return_30d " : 39 , " alpha_30d " : 39 } . get ( target , 39 )
folds = _walk_forward_folds ( rows , n_folds , embargo_days = embargo_days )
if not folds :
return { * * out , " err " : " fold 구성 실패 (표본 부족) " }
# ── 1. Linear (Ridge) ──────── ─────────────────────────
# ── 1. Linear (Ridge + 표준화) ─────────────────────────
# 피처 스케일이 -100~수천(점수 vs Amihud illiq vs log_mcap)으로 천차만별이라
# Ridge L2 패널티가 스케일에 휘둘려 계수가 왜곡됨 → 폴드마다 train에만 fit한
# StandardScaler를 파이프라인으로 결합. (scaler는 train 통계만 사용 → 누수 없음)
fold_metrics_lin = [ ]
last_lin = None
for tr , te in folds :
X_tr = np . array ( [ _row_features ( r ) for r in tr ] )
y_tr = np . array ( [ float ( r [ target ] ) for r in tr ] )
X_te = np . array ( [ _row_features ( r ) for r in te ] )
y_te = np . array ( [ float ( r [ target ] ) for r in te ] )
m = Ridge ( alpha = 1.0 ) . fit ( X_tr , y_tr )
y_p = m . predict ( X_te )
fold_metrics_lin . append ( _eval_metrics ( y_te , y_p ) )
last_lin = m # 마지막(최대 train) 모델 저장용
m = make_pipeline ( StandardScaler ( ) , Ridge ( alpha = 1.0 ) ) . fit ( X_tr , y_tr )
fold_metrics_lin . append ( _eval_metrics ( y_te , m . predict ( X_te ) ) )
linear_metrics = _aggregate_fold_metrics ( fold_metrics_lin )
# 전체 데이터 재학습 (배포용 모델)
# 전체 데이터 재학습 (배포용 모델) — 표준화 공간에서 학습 후 계수를 원본 피처 공간으로
# 역변환해 저장. /predict-price가 원본 피처로 intercept+Σcoef·x를 그대로 계산하므로
# 스케일러를 따로 들고 다닐 필요 없이 동일 예측이 보장됨.
X_all = np . array ( [ _row_features ( r ) for r in rows ] )
y_all = np . array ( [ float ( r [ target ] ) for r in rows ] )
final_lin = Ridge ( alpha = 1.0 ) . fit ( X_all , y_all )
lin_coef = { fn : round ( float ( c ) , 6 ) for fn , c in zip ( LEARN_FEATURE_NAMES , final_lin . coef_ ) }
final_pipe = make_pipeline ( StandardScaler ( ) , Ridge ( alpha = 1.0 ) ) . fit ( X_all , y_all )
_scaler = final_pipe . named_steps [ " standardscaler " ]
_ridge = final_pipe . named_steps [ " ridge " ]
_scale = np . where ( _scaler . scale_ == 0 , 1.0 , _scaler . scale_ )
raw_coef = _ridge . coef_ / _scale
lin_intercept = float ( _ridge . intercept_ - np . sum ( _ridge . coef_ * _scaler . mean_ / _scale ) )
lin_coef = { fn : round ( float ( c ) , 6 ) for fn , c in zip ( LEARN_FEATURE_NAMES , raw_coef ) }
# ── 2. Random Forest ──────────────────────────────────
fold_metrics_rf = [ ]
@@ -4190,7 +4701,7 @@ async def learn_pricing(days: int = Query(default=180, ge=14, le=730),
sample_size = $ 10 , period_days = $ 11
""" , today_d, segment, target,
json . dumps ( LEARN_FEATURE_NAMES ) , json . dumps ( lin_coef ) ,
float ( final_lin . intercept_ ) ,
lin_intercept ,
linear_metrics . get ( " r2_oos " ) , linear_metrics . get ( " ic_spearman " ) ,
linear_metrics . get ( " hit_ratio " ) , len ( rows ) , days )
# RF
@@ -5475,13 +5986,16 @@ async def ensure_trading_tables():
# 자동매매 안전 설정 — 보수적 기본값
TRADE_SETTINGS = {
" enabled " : True , # 자동매매 ON/OFF 토글
" auto_execute " : True , # 제안 즉시 자동 체결(모의). False면 텔레그램 버튼 승인 대기
" max_position_pct " : 10.0 , # 종목당 자본 ≤ 10%
" daily_loss_limit_pct " : - 3.0 , # 일일 손실 -3% 도달 시 매매 중단
" max_orders_per_day " : 10 ,
" min_conviction " : 4 , # Gemini 확신도 ≥4
" require_agreement " : True , # 두 LLM 일치 필수
" min_diagnosis_score " : 3 , # 5가지 진단 ≥3
" default_capital " : 10_000_000 ,
# 자본금: .env TRADE_DEFAULT_CAPITAL로 설정(미설정 시 1천만). 손실%·포지션사이즈 기준값.
# 런타임 변경은 POST /trade/settings?default_capital=... (재시작 시 env 값으로 복귀).
" default_capital " : int ( os . getenv ( " TRADE_DEFAULT_CAPITAL " , " 10000000 " ) ) ,
# 매도 신호 임계값
" stop_loss_pct " : - 8.0 ,
" take_profit_pct " : 15.0 ,
@@ -5635,10 +6149,12 @@ async def propose_buy_order(conn, code: str, capital: int = None) -> dict:
@app.post ( " /trade/propose/ {code} " )
async def trade_propose ( code : str ,
capital : int = Query ( default = 10_000_000 , ge = 1_000_000 ) ) :
""" 수동 매수 신호 평가 + confirm 요청 (자동 cron으로도 호출됨). """
capital : int | None = Query ( default = None , ge = 1_000_000 ) ) :
""" 수동 매수 신호 평가 + confirm 요청 (자동 cron으로도 호출됨).
capital 미지정 시 설정 자본금 ( TRADE_SETTINGS [ ' default_capital ' ] ) 사용 . """
cap = capital or TRADE_SETTINGS [ " default_capital " ]
async with pg_pool . acquire ( ) as conn :
return await propose_buy_order ( conn , code , capital )
return await propose_buy_order ( conn , code , cap )
@app.post ( " /trade/cancel/ {order_id} " )
@@ -5671,9 +6187,11 @@ async def trade_orders_list(status: str = Query(default=""),
@app.post ( " /trade/scan " )
async def trade_scan ( capital : int = Query ( default = 10_000_000 ) ,
async def trade_scan ( capital : int | None = Query ( default = None ) ,
auto_propose : bool = Query ( default = False ) ) :
""" 모든 강력매수 종목 평가 → 통과 종목 표시 (또는 자동 proposal). """
""" 모든 강력매수 종목 평가 → 통과 종목 표시 (또는 자동 proposal).
capital 미지정 시 설정 자본금 ( TRADE_SETTINGS [ ' default_capital ' ] ) 사용 . """
cap = capital or TRADE_SETTINGS [ " default_capital " ]
async with pg_pool . acquire ( ) as conn :
codes = [ r [ " stock_code " ] for r in await conn . fetch ( """
SELECT stock_code FROM stock_scores
@@ -5686,7 +6204,7 @@ async def trade_scan(capital: int = Query(default=10_000_000),
sig = await evaluate_buy_signal ( conn , c )
results . append ( { " code " : c , * * sig } )
if auto_propose and sig [ " ok " ] :
await propose_buy_order ( conn , c , capital )
await propose_buy_order ( conn , c , cap )
return { " evaluated " : len ( results ) , " results " : results }
@@ -5798,10 +6316,8 @@ async def trade_sell_scan():
return { " checked " : len ( codes ) , " proposed " : len ( results ) , " results " : results }
@app.post ( " /trade/confirm/ {order_id} " )
async def trade_confirm_v2 ( order_id : int ) : # noqa: F811 (이전 정의 override)
""" 매수/매도 confirm 통합 처리. """
async with pg_pool . acquire ( ) as conn :
async def _fill_order ( conn , order_id : int , auto : bool = False ) - > dict :
""" 주문 체결(모의=DB 기록, 실제 브로커 전송 없음). auto=True면 자동매매 체결. """
order = await conn . fetchrow (
" SELECT * FROM trading_orders WHERE id=$1 AND status= ' pending ' " , order_id )
if not order :
@@ -5810,17 +6326,17 @@ async def trade_confirm_v2(order_id: int): # noqa: F811 (이전 정의 overri
await conn . execute (
" UPDATE trading_orders SET status= ' expired ' WHERE id=$1 " , order_id )
return { " error " : " 주문 만료 " }
tag = " 자동체결 " if auto else " 체결 "
if order [ " side " ] == " buy " :
await conn . execute ( """
INSERT INTO user_portfolio ( stock_code , stock_name , buy_price , qty , memo )
VALUES ( $ 1 , $ 2 , $ 3 , $ 4 , $ 5 )
""" , order[ " stock_code " ], order[ " stock_name " ], order[ " price " ],
order [ " qty " ] , f " 자동매매 # { order_id } (모의) " )
msg = ( f " ✅ <b>매수 체결 # { order_id } </b> \n "
msg = ( f " ✅ <b>매수 { tag } # { order_id } </b> (모의) \n "
f " { order [ ' stock_name ' ] } ( { order [ ' stock_code ' ] } ) "
f " { order [ ' qty ' ] } 주 × { order [ ' price ' ] : , } 원 " )
else : # sell
# 보유 종목 비활성화 + 실현손익 계산
held = await conn . fetchrow ( """
SELECT id , buy_price , qty FROM user_portfolio
WHERE stock_code = $ 1 AND active = true LIMIT 1
@@ -5830,7 +6346,6 @@ async def trade_confirm_v2(order_id: int): # noqa: F811 (이전 정의 overri
realized = ( order [ " price " ] - held [ " buy_price " ] ) * order [ " qty " ]
await conn . execute (
" UPDATE user_portfolio SET active=false WHERE id=$1 " , held [ " id " ] )
# 일일 손익 누적
await conn . execute ( """
INSERT INTO trading_daily_pnl ( dt , realized_pnl , trades_count )
VALUES ( CURRENT_DATE , $ 1 , 1 )
@@ -5840,7 +6355,7 @@ async def trade_confirm_v2(order_id: int): # noqa: F811 (이전 정의 overri
updated_at = NOW ( )
""" , realized)
sign = " 🟢 " if realized > 0 else " 🔴 "
msg = ( f " { sign } <b>매도 체결 # { order_id } </b> \n "
msg = ( f " { sign } <b>매도 { tag } # { order_id } </b> (모의) \n "
f " { order [ ' stock_name ' ] } ( { order [ ' stock_code ' ] } ) "
f " { order [ ' qty ' ] } 주 × { order [ ' price ' ] : , } 원 \n "
f " 실현손익: { realized : +, } 원 " )
@@ -5852,6 +6367,51 @@ async def trade_confirm_v2(order_id: int): # noqa: F811 (이전 정의 overri
await send_telegram ( msg )
return { " status " : " filled " , " order_id " : order_id }
@app.post ( " /trade/confirm/ {order_id} " )
async def trade_confirm_v2 ( order_id : int ) : # noqa: F811 (이전 정의 override)
""" 매수/매도 confirm 통합 처리 (수동 버튼). """
async with pg_pool . acquire ( ) as conn :
return await _fill_order ( conn , order_id , auto = False )
@app.get ( " /trade/history " )
async def trade_history ( days : int = Query ( default = 30 , ge = 1 , le = 365 ) ) :
""" 모의매매 이력 + 손익 — 체결내역·일별손익·보유 평가손익. """
async with pg_pool . acquire ( ) as conn :
fills = await conn . fetch ( """
SELECT stock_code , stock_name , side , qty , price , filled_at
FROM trading_orders
WHERE status = ' filled ' AND filled_at > = CURRENT_DATE - ( $ 1 : : int )
ORDER BY filled_at DESC
""" , days)
daily = await conn . fetch ( """
SELECT dt , realized_pnl , unrealized_pnl , trades_count , halted
FROM trading_daily_pnl WHERE dt > = CURRENT_DATE - ( $ 1 : : int ) ORDER BY dt DESC
""" , days)
holds = await conn . fetch ( """
SELECT p . stock_code , p . stock_name , p . buy_price , p . qty ,
( SELECT price FROM stock_technical WHERE stock_code = p . stock_code ) cur
FROM user_portfolio p WHERE active = true
""" )
realized = sum ( ( r [ " realized_pnl " ] or 0 ) for r in daily )
holdings , unreal = [ ] , 0
for h in holds :
cur = h [ " cur " ] or h [ " buy_price " ]
pl = int ( ( cur - h [ " buy_price " ] ) * h [ " qty " ] ) if h [ " buy_price " ] else 0
unreal + = pl
holdings . append ( { " code " : h [ " stock_code " ] , " name " : h [ " stock_name " ] ,
" buy_price " : h [ " buy_price " ] , " qty " : h [ " qty " ] , " cur_price " : cur ,
" pl " : pl , " pl_pct " : round ( ( cur / h [ " buy_price " ] - 1 ) * 100 , 1 ) if h [ " buy_price " ] else 0 } )
return {
" realized_pnl " : realized , " unrealized_pnl " : unreal , " total_pnl " : realized + unreal ,
" fill_count " : len ( fills ) ,
" holdings " : holdings ,
" daily_pnl " : [ { " dt " : str ( r [ " dt " ] ) , " realized " : r [ " realized_pnl " ] ,
" unrealized " : r [ " unrealized_pnl " ] , " trades " : r [ " trades_count " ] ,
" halted " : r [ " halted " ] } for r in daily ] ,
" fills " : [ { " code " : f [ " stock_code " ] , " name " : f [ " stock_name " ] , " side " : f [ " side " ] ,
" qty " : f [ " qty " ] , " price " : f [ " price " ] , " at " : str ( f [ " filled_at " ] ) } for f in fills ] ,
}
async def auto_trade_scan_job ( ) :
""" 5분마다: 매수 신호 + 매도 신호 자동 스캔. 한도 도달 시 자동 halt. """
@@ -5861,16 +6421,31 @@ async def auto_trade_scan_job():
async with pg_pool . acquire ( ) as conn :
pnl = await conn . fetchrow (
" SELECT realized_pnl, halted FROM trading_daily_pnl WHERE dt=CURRENT_DATE " )
if pnl and pnl [ " halted " ] :
return # 이미 중단된 날
capital = TRADE_SETTINGS [ " default_capital " ]
if pnl and not pnl [ " halted " ] :
loss_pct = ( pnl [ " realized_pnl " ] or 0 ) / capital * 100
realized = ( pnl [ " realized_pnl " ] if pnl else 0 ) or 0
# 미실현 손익까지 합산. 기존엔 realized만 봐서 ① 평가손실이 아무리 커도
# 매도 전엔 halt가 안 걸리고 ② 매도 없는 날은 pnl row 자체가 없어 체크를 건너뛰는
# 구멍이 있었음 → 보유 평가손익을 더해 포트폴리오 기준으로 판단.
unreal = await conn . fetchval ( """
SELECT COALESCE ( SUM ( ( st . price - p . buy_price ) * p . qty ) , 0 )
FROM user_portfolio p
JOIN stock_technical st ON st . stock_code = p . stock_code
WHERE p . active = true AND p . buy_price > 0 AND st . price > 0
""" ) or 0
loss_pct = ( realized + unreal ) / capital * 100
if loss_pct < = TRADE_SETTINGS [ " daily_loss_limit_pct " ] :
await conn . execute ( """
UPDATE trading_daily_pnl SET halted = true WHERE dt = CURRENT_DATE
""" )
INSERT INTO trading_daily_pnl ( dt , unrealized_pnl , halted )
VALUES ( CURRENT_DATE , $ 1 , true )
ON CONFLICT ( dt ) DO UPDATE SET
unrealized_pnl = $ 1 , halted = true , updated_at = NOW ( )
""" , int(unreal))
await send_telegram (
f " 🛑 <b>일일 손실 한도 도달</b> \n "
f " 손실 { loss_pct : .2f } % ≤ - { abs ( TRADE_SETTINGS [ ' daily_loss_limit_pct ' ] ) } % \n "
f " 실현+평가 손실 { loss_pct : .2f } % ≤ - { abs ( TRADE_SETTINGS [ ' daily_loss_limit_pct ' ] ) } % \n "
f " (실현 { realized : +, } 원 / 평가 { int ( unreal ) : +, } 원) \n "
f " 오늘 자동매매 중단됨 "
)
return
@@ -5906,6 +6481,13 @@ async def auto_trade_scan_job():
sig = await evaluate_buy_signal ( conn , c )
if sig [ " ok " ] :
await propose_buy_order ( conn , c )
# 자동실행(모의): auto_execute면 오늘 제안된 pending 주문 즉시 체결
if TRADE_SETTINGS . get ( " auto_execute " ) :
pend = await conn . fetch (
" SELECT id FROM trading_orders WHERE status= ' pending ' "
" AND proposed_at::date=CURRENT_DATE " )
for p in pend :
await _fill_order ( conn , p [ " id " ] , auto = True )
logger . info ( " auto_trade_scan.done " )
except Exception as e :
logger . error ( " auto_trade_scan.err " , error = str ( e ) )
kyu