112 lines
3.1 KiB
Python
112 lines
3.1 KiB
Python
"""Quality Strategy (ROE, GPA, CFO)."""
|
|
from typing import List, Dict
|
|
from decimal import Decimal
|
|
from datetime import datetime
|
|
from sqlalchemy.orm import Session
|
|
import pandas as pd
|
|
|
|
from app.strategies.base import BaseStrategy
|
|
from app.utils.data_helpers import (
|
|
get_ticker_list,
|
|
get_financial_statements,
|
|
calculate_quality_factors,
|
|
get_prices_on_date
|
|
)
|
|
|
|
|
|
class QualityStrategy(BaseStrategy):
|
|
"""
|
|
우량주 투자 전략.
|
|
|
|
- ROE, GPA, CFO 세 가지 수익성 지표 기반
|
|
- 높은 수익성 종목 선정
|
|
"""
|
|
|
|
def __init__(self, config: Dict = None):
|
|
"""
|
|
초기화.
|
|
|
|
Args:
|
|
config: 전략 설정
|
|
- count: 선정 종목 수 (기본 20)
|
|
"""
|
|
super().__init__(config)
|
|
self.count = config.get('count', 20)
|
|
|
|
def select_stocks(self, rebal_date: datetime, db_session: Session) -> List[str]:
|
|
"""
|
|
종목 선정.
|
|
|
|
Args:
|
|
rebal_date: 리밸런싱 날짜
|
|
db_session: 데이터베이스 세션
|
|
|
|
Returns:
|
|
선정된 종목 코드 리스트
|
|
"""
|
|
try:
|
|
# 1. 종목 리스트 조회
|
|
ticker_list = get_ticker_list(db_session)
|
|
if ticker_list.empty:
|
|
return []
|
|
|
|
tickers = ticker_list['종목코드'].tolist()
|
|
|
|
# 2. 재무제표 데이터 조회
|
|
fs_list = get_financial_statements(db_session, tickers, rebal_date)
|
|
if fs_list.empty:
|
|
return []
|
|
|
|
# 3. 퀄리티 팩터 계산 (ROE, GPA, CFO)
|
|
quality_df = calculate_quality_factors(fs_list)
|
|
if quality_df.empty:
|
|
return []
|
|
|
|
# 4. 티커 테이블과 병합
|
|
data_bind = ticker_list[['종목코드', '종목명']].merge(
|
|
quality_df,
|
|
how='left',
|
|
on='종목코드'
|
|
)
|
|
|
|
# 5. ROE, GPA, CFO 모두 있는 종목만 필터링
|
|
data_bind = data_bind.dropna(subset=['ROE', 'GPA', 'CFO'])
|
|
|
|
if data_bind.empty:
|
|
return []
|
|
|
|
# 6. 각 지표별 순위 계산 (높을수록 좋은 지표이므로 ascending=False)
|
|
quality_rank = data_bind[['ROE', 'GPA', 'CFO']].rank(ascending=False, axis=0)
|
|
|
|
# 7. 순위 합산 후 재순위
|
|
quality_sum = quality_rank.sum(axis=1, skipna=False).rank()
|
|
|
|
# 8. 상위 N개 선정
|
|
data_bind['rank'] = quality_sum
|
|
selected = data_bind[data_bind['rank'] <= self.count]
|
|
|
|
return selected['종목코드'].tolist()
|
|
|
|
except Exception as e:
|
|
print(f"Quality 전략 종목 선정 오류: {e}")
|
|
return []
|
|
|
|
def get_prices(
|
|
self,
|
|
tickers: List[str],
|
|
date: datetime,
|
|
db_session: Session
|
|
) -> Dict[str, Decimal]:
|
|
"""
|
|
종목 가격 조회.
|
|
|
|
Args:
|
|
tickers: 종목 코드 리스트
|
|
date: 조회 날짜
|
|
db_session: 데이터베이스 세션
|
|
|
|
Returns:
|
|
{ticker: price} 딕셔너리
|
|
"""
|
|
return get_prices_on_date(db_session, tickers, date)
|