penti/backend/app/strategies/composite/super_quality.py

"""Super Quality Strategy (F-Score + GPA)."""
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.strategies.factors.f_score import FScoreStrategy
from app.utils.data_helpers import (
    get_ticker_list,
    get_financial_statements,
    get_prices_on_date
)


class SuperQualityStrategy(BaseStrategy):
    """
    슈퍼 퀄리티 전략 (F-Score + GPA).

    - F-Score 3점인 소형주 중
    - GPA (Gross Profit to Assets)가 높은 종목 선정
    """

    def __init__(self, config: Dict = None):
        """
        초기화.

        Args:
            config: 전략 설정
                - count: 선정 종목 수 (기본 20)
                - min_f_score: 최소 F-Score (기본 3)
                - size_filter: 시가총액 필터 (기본 '소형주')
        """
        super().__init__(config)
        self.count = config.get('count', 20)
        self.min_f_score = config.get('min_f_score', 3)
        self.size_filter = config.get('size_filter', '소형주')

        # F-Score 전략 인스턴스
        self.f_score_strategy = FScoreStrategy(config={
            'count': 1000,  # 많은 종목 선정 (GPA로 필터링)
            'min_score': self.min_f_score,
            'size_filter': self.size_filter
        })

    def select_stocks(self, rebal_date: datetime, db_session: Session) -> List[str]:
        """
        종목 선정.

        Args:
            rebal_date: 리밸런싱 날짜
            db_session: 데이터베이스 세션

        Returns:
            선정된 종목 코드 리스트
        """
        try:
            # 1. F-Score 계산
            f_score_df = self.f_score_strategy._calculate_f_score(rebal_date, db_session)

            if f_score_df.empty:
                return []

            # 2. F-Score 3점 & 소형주 필터
            filtered = f_score_df[
                (f_score_df['f_score'] >= self.min_f_score) &
                (f_score_df['분류'] == self.size_filter)
            ]

            if filtered.empty:
                print(f"F-Score {self.min_f_score}점 {self.size_filter} 종목 없음")
                return []

            # 3. GPA 계산
            gpa_df = self._calculate_gpa(rebal_date, db_session, filtered['종목코드'].tolist())

            if gpa_df.empty:
                return []

            # 4. GPA 병합
            result = filtered.merge(gpa_df, on='종목코드', how='left')
            result['GPA'] = result['GPA'].fillna(-1).astype(float)

            # 5. GPA 순으로 상위 N개 종목 선정
            top_stocks = result.nlargest(self.count, 'GPA')

            print(f"F-Score {self.min_f_score}점 {self.size_filter}: {len(filtered)}개")
            print(f"GPA 상위 {self.count}개 선정")

            return top_stocks['종목코드'].tolist()

        except Exception as e:
            print(f"Super Quality 종목 선정 오류: {e}")
            return []

    def _calculate_gpa(
        self,
        base_date: datetime,
        db_session: Session,
        tickers: List[str]
    ) -> pd.DataFrame:
        """
        GPA (Gross Profit to Assets) 계산.

        Args:
            base_date: 기준일
            db_session: 데이터베이스 세션
            tickers: 종목 리스트

        Returns:
            GPA DataFrame
        """
        # 재무제표 데이터 조회
        fs_list = get_financial_statements(db_session, tickers, base_date)
        if fs_list.empty:
            return pd.DataFrame()

        # 필요한 계정만 필터링
        fs_filtered = fs_list[fs_list['계정'].isin(['매출총이익', '자산'])].copy()

        if fs_filtered.empty:
            return pd.DataFrame()

        # Pivot
        fs_pivot = fs_filtered.pivot_table(
            index='종목코드',
            columns='계정',
            values='값',
            aggfunc='first'
        )

        # GPA 계산
        if '매출총이익' in fs_pivot.columns and '자산' in fs_pivot.columns:
            fs_pivot['GPA'] = fs_pivot['매출총이익'] / fs_pivot['자산']
        else:
            fs_pivot['GPA'] = None

        return fs_pivot.reset_index()[['종목코드', 'GPA']]

    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)