From 65bc4cb6236f508bab4c15e4590eb74e0e509459 Mon Sep 17 00:00:00 2001 From: zephyrdark Date: Thu, 19 Feb 2026 15:12:18 +0900 Subject: [PATCH] feat: add KJBStrategy ranking class and API endpoint Co-Authored-By: Claude Opus 4.6 --- backend/app/api/strategy.py | 19 +++++- backend/app/schemas/strategy.py | 5 ++ backend/app/services/strategy/__init__.py | 3 +- backend/app/services/strategy/kjb.py | 79 +++++++++++++++++++++++ 4 files changed, 103 insertions(+), 3 deletions(-) diff --git a/backend/app/api/strategy.py b/backend/app/api/strategy.py index f1a6331..3f04bca 100644 --- a/backend/app/api/strategy.py +++ b/backend/app/api/strategy.py @@ -7,9 +7,9 @@ from sqlalchemy.orm import Session from app.core.database import get_db from app.api.deps import CurrentUser from app.schemas.strategy import ( - MultiFactorRequest, QualityRequest, ValueMomentumRequest, StrategyResult, + MultiFactorRequest, QualityRequest, ValueMomentumRequest, KJBRequest, StrategyResult, ) -from app.services.strategy import MultiFactorStrategy, QualityStrategy, ValueMomentumStrategy +from app.services.strategy import MultiFactorStrategy, QualityStrategy, ValueMomentumStrategy, KJBStrategy router = APIRouter(prefix="/api/strategy", tags=["strategy"]) @@ -61,3 +61,18 @@ async def run_value_momentum( value_weight=request.value_weight, momentum_weight=request.momentum_weight, ) + + +@router.post("/kjb", response_model=StrategyResult) +async def run_kjb( + request: KJBRequest, + current_user: CurrentUser, + db: Session = Depends(get_db), +): + """Run KJB strategy.""" + strategy = KJBStrategy(db) + return strategy.run( + universe_filter=request.universe, + top_n=request.top_n, + base_date=request.base_date, + ) diff --git a/backend/app/schemas/strategy.py b/backend/app/schemas/strategy.py index 1646c57..d208227 100644 --- a/backend/app/schemas/strategy.py +++ b/backend/app/schemas/strategy.py @@ -50,6 +50,11 @@ class ValueMomentumRequest(StrategyRequest): momentum_weight: FloatDecimal = Field(default=Decimal("0.5"), ge=0, le=1) +class KJBRequest(StrategyRequest): + """KJB strategy request.""" + pass + + class StockFactor(BaseModel): """Factor scores for a single stock.""" ticker: str diff --git a/backend/app/services/strategy/__init__.py b/backend/app/services/strategy/__init__.py index 8c5de1c..d9aacd1 100644 --- a/backend/app/services/strategy/__init__.py +++ b/backend/app/services/strategy/__init__.py @@ -2,5 +2,6 @@ from app.services.strategy.base import BaseStrategy from app.services.strategy.multi_factor import MultiFactorStrategy from app.services.strategy.quality import QualityStrategy from app.services.strategy.value_momentum import ValueMomentumStrategy +from app.services.strategy.kjb import KJBStrategy, KJBSignalGenerator -__all__ = ["BaseStrategy", "MultiFactorStrategy", "QualityStrategy", "ValueMomentumStrategy"] +__all__ = ["BaseStrategy", "MultiFactorStrategy", "QualityStrategy", "ValueMomentumStrategy", "KJBStrategy", "KJBSignalGenerator"] diff --git a/backend/app/services/strategy/kjb.py b/backend/app/services/strategy/kjb.py index 6f61337..349727a 100644 --- a/backend/app/services/strategy/kjb.py +++ b/backend/app/services/strategy/kjb.py @@ -97,3 +97,82 @@ class KJBSignalGenerator: 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], + )