107 lines
2.9 KiB
Python
107 lines
2.9 KiB
Python
|
"""Value Strategy (PER, PBR)."""
|
||
|
|
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_value_indicators,
|
||
|
|
calculate_value_rank,
|
||
|
|
get_prices_on_date
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class ValueStrategy(BaseStrategy):
|
||
|
|
"""
|
||
|
|
가치 투자 전략.
|
||
|
|
|
||
|
|
- PER, PBR 두 가지 가치 지표 기반
|
||
|
|
- 낮은 밸류에이션 종목 선정
|
||
|
|
"""
|
||
|
|
|
||
|
|
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. PER, PBR 조회
|
||
|
|
value_list = get_value_indicators(db_session, tickers, include_psr_pcr=False)
|
||
|
|
if value_list.empty:
|
||
|
|
return []
|
||
|
|
|
||
|
|
# 3. 가로로 긴 형태로 변경 (pivot)
|
||
|
|
value_pivot = value_list.pivot(index='종목코드', columns='지표', values='값')
|
||
|
|
|
||
|
|
# 4. 티커 테이블과 가치 지표 테이블 병합
|
||
|
|
data_bind = ticker_list[['종목코드', '종목명']].merge(
|
||
|
|
value_pivot,
|
||
|
|
how='left',
|
||
|
|
on='종목코드'
|
||
|
|
)
|
||
|
|
|
||
|
|
# 5. PER, PBR 둘 다 있는 종목만 필터링
|
||
|
|
data_bind = data_bind.dropna(subset=['PER', 'PBR'])
|
||
|
|
|
||
|
|
if data_bind.empty:
|
||
|
|
return []
|
||
|
|
|
||
|
|
# 6. 순위 계산
|
||
|
|
value_sum = calculate_value_rank(data_bind.set_index('종목코드'), ['PER', 'PBR'])
|
||
|
|
|
||
|
|
# 7. 상위 N개 선정
|
||
|
|
data_bind['rank'] = value_sum
|
||
|
|
selected = data_bind[data_bind['rank'] <= self.count]
|
||
|
|
|
||
|
|
return selected['종목코드'].tolist()
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Value 전략 종목 선정 오류: {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)
|