galaxis-po/backend/app/services/strategy/kjb.py

"""
Kim Jong-bong (KJB) strategy implementation.

Signal-based short-term trading strategy:
- Universe: market cap top 30, daily trading value >= 200B KRW
- Entry: relative strength > KOSPI + breakout or large candle
- Exit: stop-loss -3%, take-profit +5%/+10%, trailing stop
"""
from datetime import date
from decimal import Decimal
from typing import Dict, List, Optional

import pandas as pd
from sqlalchemy.orm import Session

from app.services.strategy.base import BaseStrategy
from app.schemas.strategy import StockFactor, StrategyResult, UniverseFilter
from app.services.factor_calculator import FactorCalculator


class KJBSignalGenerator:
    """
    Generates daily buy/sell signals based on KJB rules.
    Pure computation - no DB access. Takes DataFrames as input.
    """

    def calculate_relative_strength(
        self,
        stock_df: pd.DataFrame,
        kospi_df: pd.DataFrame,
        lookback: int = 10,
    ) -> pd.Series:
        """
        RS = (stock return / market return) * 100
        RS > 100 means stock outperforms market.
        """
        stock_ret = stock_df["close"].pct_change(lookback)
        kospi_ret = kospi_df["close"].pct_change(lookback)

        # Align on common index
        aligned = pd.DataFrame({
            "stock_ret": stock_ret,
            "kospi_ret": kospi_ret,
        }).dropna()

        rs = pd.Series(dtype=float, index=stock_df.index)
        for idx in aligned.index:
            market_ret = aligned.loc[idx, "kospi_ret"]
            stock_r = aligned.loc[idx, "stock_ret"]
            if abs(market_ret) < 1e-10:
                rs[idx] = 100.0 if abs(stock_r) < 1e-10 else (200.0 if stock_r > 0 else 0.0)
            else:
                rs[idx] = (stock_r / market_ret) * 100
        return rs

    def detect_breakout(
        self,
        stock_df: pd.DataFrame,
        lookback: int = 20,
    ) -> pd.Series:
        """Close > highest high of previous lookback days."""
        prev_high = stock_df["high"].rolling(lookback).max().shift(1)
        return stock_df["close"] > prev_high

    def detect_large_candle(
        self,
        stock_df: pd.DataFrame,
        pct_threshold: float = 0.05,
        vol_multiplier: float = 1.5,
    ) -> pd.Series:
        """
        Daily return >= 5% AND volume >= 1.5x 20-day average.
        """
        daily_return = stock_df["close"].pct_change()
        avg_volume = stock_df["volume"].rolling(20).mean()
        volume_ratio = stock_df["volume"] / avg_volume
        return (daily_return >= pct_threshold) & (volume_ratio >= vol_multiplier)

    def generate_signals(
        self,
        stock_df: pd.DataFrame,
        kospi_df: pd.DataFrame,
        rs_lookback: int = 10,
        breakout_lookback: int = 20,
    ) -> pd.DataFrame:
        """
        Buy when: RS > 100 AND (breakout OR large candle)
        """
        rs = self.calculate_relative_strength(stock_df, kospi_df, rs_lookback)
        breakout = self.detect_breakout(stock_df, breakout_lookback)
        large_candle = self.detect_large_candle(stock_df)

        signals = pd.DataFrame(index=stock_df.index)
        signals["rs"] = rs
        signals["breakout"] = breakout.fillna(False)
        signals["large_candle"] = large_candle.fillna(False)
        signals["buy"] = (rs > 100) & (breakout.fillna(False) | large_candle.fillna(False))
        signals["buy"] = signals["buy"].fillna(False)
        return signals


class KJBStrategy(BaseStrategy):
    """
    KJB strategy for stock ranking.
    Ranks stocks by relative strength and momentum.
    Compatible with existing strategy pattern (returns StrategyResult).
    """

    strategy_name = "kjb"

    def run(
        self,
        universe_filter: UniverseFilter,
        top_n: int,
        base_date: date = None,
        **kwargs,
    ) -> StrategyResult:
        if base_date is None:
            base_date = date.today()

        # Get universe - filter to top 30 by market cap
        stocks = self.get_universe(universe_filter)
        stocks.sort(key=lambda s: s.market_cap or 0, reverse=True)
        stocks = stocks[:30]

        tickers = [s.ticker for s in stocks]
        stock_map = {s.ticker: s for s in stocks}

        if not tickers:
            return StrategyResult(
                strategy_name=self.strategy_name,
                base_date=base_date,
                universe_count=0,
                result_count=0,
                stocks=[],
            )

        # Get valuations and sectors
        valuations = self.factor_calc.get_valuations(tickers, base_date)
        sectors = self.factor_calc.get_sectors(tickers)

        # Calculate 1-month momentum as ranking proxy
        momentum = self.factor_calc.calculate_momentum(
            tickers, base_date, months=1, skip_recent=0,
        )

        # Build results
        results = []
        for ticker in tickers:
            stock = stock_map[ticker]
            val = valuations.get(ticker)
            mom = momentum.get(ticker, Decimal("0"))

            results.append(StockFactor(
                ticker=ticker,
                name=stock.name,
                market=stock.market,
                sector_name=sectors.get(ticker),
                market_cap=int(stock.market_cap / 100_000_000) if stock.market_cap else None,
                close_price=Decimal(str(stock.close_price)) if stock.close_price else None,
                per=Decimal(str(val.per)) if val and val.per else None,
                pbr=Decimal(str(val.pbr)) if val and val.pbr else None,
                dividend_yield=Decimal(str(val.dividend_yield)) if val and val.dividend_yield else None,
                momentum_score=mom,
                total_score=mom,
            ))

        results.sort(key=lambda x: x.total_score or Decimal("0"), reverse=True)
        for i, r in enumerate(results[:top_n], 1):
            r.rank = i

        return StrategyResult(
            strategy_name=self.strategy_name,
            base_date=base_date,
            universe_count=len(stocks),
            result_count=min(top_n, len(results)),
            stocks=results[:top_n],
        )