""" 스크리닝 드라이런 스크립트. 최근 1달치 각 영업일에 대해 시장 상태 판단 + 거래대금 종목 스크리닝 실행. KIS API / 거래 기능 없음. 사용법: cd backend uv run python scripts/screening_dryrun.py uv run python scripts/screening_dryrun.py --days 20 """ import sys import argparse from pathlib import Path import importlib.util import logging from datetime import date, timedelta from typing import List, Dict import pandas as pd from pykrx import stock as pykrx_stock # kospi_screener를 직접 로드 (app 모듈 체인의 DB 의존성 우회) _screener_path = Path(__file__).parent.parent / "app" / "services" / "strategy" / "kospi_screener.py" _spec = importlib.util.spec_from_file_location("kospi_screener", _screener_path) _mod = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(_mod) KOSPIMarketStateDetector = _mod.KOSPIMarketStateDetector VolumeScreener = _mod.VolumeScreener StockInfo = _mod.StockInfo ScreeningResult = _mod.ScreeningResult MarketState = _mod.MarketState logging.basicConfig(level=logging.WARNING) logger = logging.getLogger(__name__) def get_business_days(start_date: date, end_date: date) -> List[date]: """영업일 목록 반환 (pykrx 기준).""" days = [] current = start_date while current <= end_date: if current.weekday() < 5: days.append(current) current += timedelta(days=1) return days def get_sector_map(biz_day_str: str) -> Dict[str, str]: """WISEindex 섹터 맵 반환.""" import requests SECTOR_CODES = ["G25", "G35", "G50", "G40", "G10", "G20", "G55", "G30", "G15", "G45"] sector_map = {} for code in SECTOR_CODES: try: url = f"http://www.wiseindex.com/Index/GetIndexComponets?ceil_yn=0&dt={biz_day_str}&sec_cd={code}" resp = requests.get(url, timeout=5) data = resp.json() for item in data.get("list", []): sector_map[item["CMP_CD"]] = item["SEC_NM_KOR"] except Exception: pass return sector_map def run_dryrun(days: int = 30) -> pd.DataFrame: end_date = date.today() start_date = end_date - timedelta(days=days + 30) end_str = end_date.strftime("%Y%m%d") start_str = start_date.strftime("%Y%m%d") print(f"데이터 수집 중... ({start_str} ~ {end_str})") print("KOSPI 지수 데이터 수집...") kospi_df = pykrx_stock.get_index_ohlcv(start_str, end_str, "1001") kospi_df.index.name = "date" col_count = len(kospi_df.columns) if col_count >= 6: kospi_df.columns = ["open", "high", "low", "close", "volume", "trading_value"] + list(kospi_df.columns[6:]) elif col_count >= 5: kospi_df.columns = ["open", "high", "low", "close", "volume"] + list(kospi_df.columns[5:]) target_days_raw = get_business_days(end_date - timedelta(days=days), end_date) kospi_idx_dates = set(kospi_df.index.date) target_days = [d for d in target_days_raw if d in kospi_idx_dates] print(f"분석 대상: {len(target_days)}일") if not target_days: print("분석 가능한 영업일 없음") return pd.DataFrame() latest_biz = target_days[-1].strftime("%Y%m%d") print(f"KOSPI 종목 목록 수집 ({latest_biz})...") tickers = pykrx_stock.get_market_ticker_list(latest_biz, market="KOSPI") print(f" 총 {len(tickers)}개 종목") print("섹터 정보 수집...") sector_map = get_sector_map(latest_biz) print("시가총액 수집...") try: cap_df = pykrx_stock.get_market_cap_by_ticker(latest_biz) except Exception: cap_df = pd.DataFrame() print("종목 가격 데이터 수집 중... (시간이 걸릴 수 있습니다)") data_start = (target_days[0] - timedelta(days=400)).strftime("%Y%m%d") data_end = target_days[-1].strftime("%Y%m%d") price_data: Dict[str, pd.DataFrame] = {} stock_info: Dict[str, StockInfo] = {} for i, ticker in enumerate(tickers): if i % 100 == 0: print(f" {i}/{len(tickers)} 처리 중...") try: df = pykrx_stock.get_market_ohlcv(data_start, data_end, ticker) if df is None or df.empty: continue df.index.name = "date" col_count = len(df.columns) if col_count >= 7: df.columns = ["open", "high", "low", "close", "volume", "trading_value", "change"] + list(df.columns[7:]) elif col_count >= 6: df.columns = ["open", "high", "low", "close", "volume", "trading_value"] + list(df.columns[6:]) else: df.columns = ["open", "high", "low", "close", "volume"] + list(df.columns[5:]) price_data[ticker] = df market_cap = 0 if ticker in cap_df.index: try: market_cap = int(cap_df.at[ticker, "시가총액"]) except Exception: pass name = pykrx_stock.get_market_ticker_name(ticker) sector = sector_map.get(ticker, "기타") stock_info[ticker] = StockInfo( ticker=ticker, name=name, market_cap=market_cap, sector=sector, ) except Exception as e: logger.debug("Failed %s: %s", ticker, e) continue print(f" 가격 데이터 수집 완료: {len(price_data)}개 종목") detector = KOSPIMarketStateDetector() screener = VolumeScreener() all_rows = [] for target_date in target_days: target_ts = pd.Timestamp(target_date) kospi_up_to = kospi_df[kospi_df.index <= target_ts] if len(kospi_up_to) < 20: continue market_state = detector.detect(kospi_up_to) if market_state == MarketState.crash: all_rows.append({ "날짜": target_date.strftime("%Y-%m-%d"), "시장상태": "급락(신규진입금지)", "종목코드": "-", "종목명": "-", "섹터": "-", "종가": 0, "등락률(%)": 0, "거래대금(억)": 0, "상한가": "-", "시총(억)": 0, "희소성(횟수)": 0, "신호강도": 0, }) continue filtered_price_data = {} for ticker, df in price_data.items(): df_up_to = df[df.index <= target_ts] if not df_up_to.empty and target_ts in df_up_to.index: filtered_price_data[ticker] = df_up_to results: List[ScreeningResult] = screener.screen( target_date, filtered_price_data, stock_info, ) if not results: all_rows.append({ "날짜": target_date.strftime("%Y-%m-%d"), "시장상태": market_state.value, "종목코드": "(해당없음)", "종목명": "-", "섹터": "-", "종가": 0, "등락률(%)": 0, "거래대금(억)": 0, "상한가": "-", "시총(억)": 0, "희소성(횟수)": 0, "신호강도": 0, }) else: for r in results: all_rows.append({ "날짜": target_date.strftime("%Y-%m-%d"), "시장상태": market_state.value, "종목코드": r.ticker, "종목명": r.name, "섹터": r.sector, "종가": int(r.close_price), "등락률(%)": round(r.daily_return * 100, 2), "거래대금(억)": round(r.trading_value / 1e8, 0), "상한가": "✅" if r.is_limit_up else "", "시총(억)": round(r.market_cap / 1e8, 0) if r.market_cap else 0, "희소성(횟수)": r.frequency_count, "신호강도": round(r.signal_strength, 2), }) df_result = pd.DataFrame(all_rows) return df_result def main(): parser = argparse.ArgumentParser(description="KOSPI 종목발굴 전략 드라이런") parser.add_argument("--days", type=int, default=30, help="최근 N일 (기본 30)") args = parser.parse_args() df = run_dryrun(days=args.days) if df.empty: print("결과 없음") return print("\n" + "=" * 80) print(f"김종봉식 KOSPI 종목발굴 전략 드라이런 결과 (최근 {args.days}일)") print("=" * 80) state_counts = df[df["종목코드"] != "(해당없음)"]["시장상태"].value_counts() print("\n[시장 상태 분포]") for state, count in state_counts.items(): print(f" {state}: {count}일") signal_df = df[~df["종목코드"].isin(["-", "(해당없음)"])].copy() print(f"\n[신호 발생 요약]") print(f" 총 신호 발생: {len(signal_df)}건 ({signal_df['날짜'].nunique()}일)") if not signal_df.empty: print(f"\n[전체 결과]") pd.set_option("display.max_rows", 200) pd.set_option("display.width", 120) pd.set_option("display.max_columns", 12) print(signal_df.to_string(index=False)) output_path = Path(__file__).parent / "screening_dryrun_result.csv" df.to_csv(output_path, index=False, encoding="utf-8-sig") print(f"\n결과 저장: {output_path}") if __name__ == "__main__": main()