diff --git a/backend/app/services/strategy/kjb.py b/backend/app/services/strategy/kjb.py new file mode 100644 index 0000000..6f61337 --- /dev/null +++ b/backend/app/services/strategy/kjb.py @@ -0,0 +1,99 @@ +""" +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 diff --git a/backend/tests/unit/test_kjb_signal.py b/backend/tests/unit/test_kjb_signal.py new file mode 100644 index 0000000..26d1aed --- /dev/null +++ b/backend/tests/unit/test_kjb_signal.py @@ -0,0 +1,100 @@ +""" +Unit tests for KJB signal generator. +""" +import pandas as pd +from datetime import date, timedelta + +from app.services.strategy.kjb import KJBSignalGenerator + + +def _make_price_df(closes, volumes=None, start_date=date(2024, 1, 2)): + dates = pd.bdate_range(start=start_date, periods=len(closes)) + if volumes is None: + volumes = [1000000] * len(closes) + highs = [c * 1.01 for c in closes] + lows = [c * 0.99 for c in closes] + opens = closes.copy() + return pd.DataFrame({ + "open": opens, + "high": highs, + "low": lows, + "close": closes, + "volume": volumes, + }, index=dates) + + +def _make_kospi_df(closes, start_date=date(2024, 1, 2)): + dates = pd.bdate_range(start=start_date, periods=len(closes)) + return pd.DataFrame({"close": closes}, index=dates) + + +def test_relative_strength_above_market(): + gen = KJBSignalGenerator() + stock_closes = [100 + i for i in range(25)] + kospi_closes = [100 + i * 0.5 for i in range(25)] + stock_df = _make_price_df(stock_closes) + kospi_df = _make_kospi_df(kospi_closes) + rs = gen.calculate_relative_strength(stock_df, kospi_df, lookback=10) + assert rs.dropna().iloc[-1] > 100 + + +def test_relative_strength_below_market(): + gen = KJBSignalGenerator() + stock_closes = [100 + i * 0.3 for i in range(25)] + kospi_closes = [100 + i for i in range(25)] + stock_df = _make_price_df(stock_closes) + kospi_df = _make_kospi_df(kospi_closes) + rs = gen.calculate_relative_strength(stock_df, kospi_df, lookback=10) + assert rs.dropna().iloc[-1] < 100 + + +def test_detect_breakout(): + gen = KJBSignalGenerator() + closes = [100.0] * 20 + [105.0] + stock_df = _make_price_df(closes) + breakouts = gen.detect_breakout(stock_df, lookback=20) + assert breakouts.iloc[-1] == True + assert breakouts.iloc[-2] == False + + +def test_detect_large_candle(): + gen = KJBSignalGenerator() + closes = [100.0] * 21 + [106.0] + volumes = [1000000] * 21 + [3000000] + stock_df = _make_price_df(closes, volumes) + large = gen.detect_large_candle(stock_df, pct_threshold=0.05, vol_multiplier=1.5) + assert large.iloc[-1] == True + assert large.iloc[-2] == False + + +def test_no_large_candle_low_volume(): + gen = KJBSignalGenerator() + closes = [100.0] * 21 + [106.0] + volumes = [1000000] * 22 + stock_df = _make_price_df(closes, volumes) + large = gen.detect_large_candle(stock_df) + assert large.iloc[-1] == False + + +def test_generate_buy_signal(): + gen = KJBSignalGenerator() + closes = [100.0] * 20 + [106.0] + volumes = [1000000] * 20 + [3000000] + kospi_closes = [100.0 + i * 0.1 for i in range(21)] + + stock_df = _make_price_df(closes, volumes) + kospi_df = _make_kospi_df(kospi_closes) + + signals = gen.generate_signals(stock_df, kospi_df) + assert signals["buy"].iloc[-1] == True + + +def test_no_signal_weak_stock(): + gen = KJBSignalGenerator() + # Stock underperforms market + closes = [100.0 + i * 0.1 for i in range(25)] + kospi_closes = [100 + i for i in range(25)] + stock_df = _make_price_df(closes) + kospi_df = _make_kospi_df(kospi_closes) + signals = gen.generate_signals(stock_df, kospi_df) + assert signals["buy"].iloc[-1] == False