100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
"""
|
|
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
|