""" 종목 유니버스 관리 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 []