From da4732f1e97bb8e6a3f98270070a29389bc29008 Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Fri, 6 Mar 2026 21:53:59 +0900 Subject: [PATCH] feat: Add stock universe loader for bulk backtest --- backend/app/services/stock_universe.py | 280 +++++++++++++++++++++++++ 1 file changed, 280 insertions(+) create mode 100644 backend/app/services/stock_universe.py diff --git a/backend/app/services/stock_universe.py b/backend/app/services/stock_universe.py new file mode 100644 index 0000000..0867d5c --- /dev/null +++ b/backend/app/services/stock_universe.py @@ -0,0 +1,280 @@ +""" +종목 유니버스 관리 + +KOSPI/KOSDAQ 전체 종목 리스트 조회 및 필터링 +""" + +import logging +from typing import List, Dict, Any, Optional +from datetime import date, datetime, timedelta +from dataclasses import dataclass + +import pandas as pd +from sqlalchemy import select, and_, func +from sqlalchemy.orm import selectinload + +from app.core.database import AsyncSessionLocal +from app.models.models import Stock, StockDailyPrice + +logger = logging.getLogger(__name__) + + +@dataclass +class StockFilter: + """종목 필터 조건""" + market: Optional[str] = None # 'KOSPI' | 'KOSDAQ' | None (전체) + min_market_cap: Optional[float] = None # 최소 시가총액 (억원) + min_volume: Optional[float] = None # 최소 일평균 거래량 (주) + min_price: Optional[float] = None # 최소 주가 (원) + max_price: Optional[float] = None # 최대 주가 (원) + exclude_admin: bool = True # 관리종목 제외 + sectors: Optional[List[str]] = None # 업종 필터 + top_n: Optional[int] = None # 상위 N개 종목만 + + +class StockUniverse: + """종목 유니버스 관리자""" + + @staticmethod + async def get_all_stocks( + market: Optional[str] = None, + active_only: bool = True + ) -> List[Dict[str, Any]]: + """DB에서 전체 종목 리스트 조회 + + Args: + market: 'KOSPI' | 'KOSDAQ' | None + active_only: 활성 종목만 조회 + + Returns: + [{'code': '005930', 'name': '삼성전자', 'market': 'KOSPI'}, ...] + """ + async with AsyncSessionLocal() as session: + stmt = select(Stock) + + if active_only: + stmt = stmt.where(Stock.is_active == True) + + if market: + stmt = stmt.where(Stock.market == market) + + result = await session.execute(stmt) + stocks = result.scalars().all() + + return [ + { + 'code': s.stock_code, + 'name': s.stock_name, + 'market': s.market, + 'sector': getattr(s, 'sector', None), + } + for s in stocks + ] + + @staticmethod + async def get_filtered_stocks( + filter_config: StockFilter, + reference_date: Optional[date] = None + ) -> List[str]: + """필터 조건에 맞는 종목 코드 리스트 + + Args: + filter_config: 필터 설정 + reference_date: 기준일 (None이면 최근 영업일) + + Returns: + ['005930', '000660', ...] + """ + if reference_date is None: + reference_date = datetime.now().date() - timedelta(days=1) + + async with AsyncSessionLocal() as session: + # 1. 기본 종목 필터 + stmt = select(Stock).where(Stock.is_active == True) + + if filter_config.market: + stmt = stmt.where(Stock.market == filter_config.market) + + if filter_config.exclude_admin: + # 종목명에 '관리' 포함 제외 + stmt = stmt.where(~Stock.stock_name.like('%관리%')) + + result = await session.execute(stmt) + stocks = result.scalars().all() + stock_codes = [s.stock_code for s in stocks] + + if not stock_codes: + return [] + + # 2. 가격/거래량 필터 (최근 20일 평균) + if any([filter_config.min_volume, filter_config.min_price, filter_config.max_price]): + start_date = reference_date - timedelta(days=30) + + # 종목별 평균 계산 + stmt = select( + StockDailyPrice.stock_code, + func.avg(StockDailyPrice.close_price).label('avg_price'), + func.avg(StockDailyPrice.volume).label('avg_volume') + ).where( + and_( + StockDailyPrice.stock_code.in_(stock_codes), + StockDailyPrice.trade_date >= start_date, + StockDailyPrice.trade_date <= reference_date + ) + ).group_by(StockDailyPrice.stock_code) + + result = await session.execute(stmt) + rows = result.all() + + filtered_codes = [] + for row in rows: + avg_price = float(row.avg_price) if row.avg_price else 0 + avg_volume = float(row.avg_volume) if row.avg_volume else 0 + + # 가격 필터 + if filter_config.min_price and avg_price < filter_config.min_price: + continue + if filter_config.max_price and avg_price > filter_config.max_price: + continue + + # 거래량 필터 + if filter_config.min_volume and avg_volume < filter_config.min_volume: + continue + + filtered_codes.append(row.stock_code) + + stock_codes = filtered_codes + + # 3. Top N 제한 + if filter_config.top_n and len(stock_codes) > filter_config.top_n: + stock_codes = stock_codes[:filter_config.top_n] + + logger.info(f"필터링 완료: {len(stock_codes)}개 종목") + return stock_codes + + @staticmethod + async def get_liquid_stocks( + market: str = 'KOSPI', + top_n: int = 100 + ) -> List[str]: + """거래대금 상위 N개 종목 (유동성 높은 종목) + + Args: + market: 'KOSPI' | 'KOSDAQ' + top_n: 상위 N개 + + Returns: + 종목 코드 리스트 + """ + filter_config = StockFilter( + market=market, + min_volume=100000, # 일 10만주 이상 + min_price=5000, # 5천원 이상 + exclude_admin=True, + top_n=top_n + ) + return await StockUniverse.get_filtered_stocks(filter_config) + + @staticmethod + async def get_kospi200_like( + top_n: int = 200 + ) -> List[str]: + """KOSPI200 유사 종목 (시총 상위 200) + + 실제 KOSPI200 구성종목 데이터가 없으므로 시총 상위로 근사 + """ + filter_config = StockFilter( + market='KOSPI', + min_market_cap=5000, # 5천억 이상 + min_volume=50000, + exclude_admin=True, + top_n=top_n + ) + return await StockUniverse.get_filtered_stocks(filter_config) + + @staticmethod + def get_preset_universes() -> Dict[str, StockFilter]: + """사전 정의된 유니버스 프리셋 + + Returns: + {'large_cap': StockFilter(...), ...} + """ + return { + 'all': StockFilter( + exclude_admin=True + ), + 'kospi_all': StockFilter( + market='KOSPI', + exclude_admin=True + ), + 'kosdaq_all': StockFilter( + market='KOSDAQ', + exclude_admin=True + ), + 'large_cap': StockFilter( + market='KOSPI', + min_market_cap=10000, # 1조 이상 + min_volume=100000, + exclude_admin=True, + top_n=100 + ), + 'mid_cap': StockFilter( + market='KOSPI', + min_market_cap=3000, # 3천억 이상 + max_price=100000, # 10만원 이하 + min_volume=50000, + exclude_admin=True, + top_n=200 + ), + 'liquid_kosdaq': StockFilter( + market='KOSDAQ', + min_volume=200000, # 20만주 이상 + min_price=3000, # 3천원 이상 + exclude_admin=True, + top_n=100 + ), + 'penny_stocks': StockFilter( + min_price=1000, + max_price=5000, # 1천~5천원 + min_volume=500000, # 거래량 많은 것만 + exclude_admin=True, + top_n=50 + ), + } + + +# ── FinanceDataReader 연동 (API 없이 전체 종목 조회) ───────────────────────────────── + +async def get_all_krx_stocks_from_fdr() -> List[Dict[str, Any]]: + """FinanceDataReader로 KRX 전체 종목 조회 + + DB에 없어도 사용 가능! + + Returns: + [{'code': '005930', 'name': '삼성전자', 'market': 'KOSPI'}, ...] + """ + try: + import FinanceDataReader as fdr + + # KRX 전체 종목 + df_krx = fdr.StockListing('KRX') + + stocks = [] + for _, row in df_krx.iterrows(): + stocks.append({ + 'code': row['Code'], + 'name': row['Name'], + 'market': row['Market'], + 'sector': row.get('Sector', None), + 'listing_date': row.get('ListingDate', None), + }) + + logger.info(f"FDR에서 {len(stocks)}개 종목 조회") + return stocks + + except ImportError: + logger.error("FinanceDataReader 미설치. pip install finance-datareader") + return [] + except Exception as e: + logger.error(f"FDR 조회 실패: {e}") + return []