diff --git a/docs/plans/2026-02-03-phase4-quant-strategy.md b/docs/plans/2026-02-03-phase4-quant-strategy.md new file mode 100644 index 0000000..1a08a5d --- /dev/null +++ b/docs/plans/2026-02-03-phase4-quant-strategy.md @@ -0,0 +1,2032 @@ +# Phase 4: Quant Strategy Implementation Plan + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** Implement quant strategy engine with multi-factor, quality, and value-momentum strategies, plus stock screening and market data APIs. + +**Architecture:** Factor-based stock ranking system. FactorCalculator computes value/quality/momentum factors from DB data. Strategy services combine factors with configurable weights. Screener filters universe and ranks by strategy scores. + +**Tech Stack:** FastAPI, SQLAlchemy, Pydantic, pandas, NumPy, Next.js, React, TypeScript + +--- + +## Task 1: Strategy Pydantic Schemas + +**Files:** +- Create: `backend/app/schemas/strategy.py` +- Update: `backend/app/schemas/__init__.py` + +**Step 1: Create strategy schemas** + +```python +""" +Quant strategy related Pydantic schemas. +""" +from datetime import date +from decimal import Decimal +from typing import Optional, List, Dict + +from pydantic import BaseModel, Field + + +class FactorWeights(BaseModel): + """Factor weights for multi-factor strategy.""" + value: Decimal = Field(default=Decimal("0.25"), ge=0, le=1) + quality: Decimal = Field(default=Decimal("0.25"), ge=0, le=1) + momentum: Decimal = Field(default=Decimal("0.25"), ge=0, le=1) + low_vol: Decimal = Field(default=Decimal("0.25"), ge=0, le=1) + + +class UniverseFilter(BaseModel): + """Stock universe filtering criteria.""" + markets: List[str] = ["KOSPI", "KOSDAQ"] + min_market_cap: Optional[int] = None # in 억원 + max_market_cap: Optional[int] = None + exclude_stock_types: List[str] = ["spac", "preferred", "reit"] + exclude_sectors: List[str] = [] + + +class StrategyRequest(BaseModel): + """Base request for running a strategy.""" + universe: UniverseFilter = UniverseFilter() + top_n: int = Field(default=30, ge=1, le=100) + base_date: Optional[date] = None + + +class MultiFactorRequest(StrategyRequest): + """Multi-factor strategy request.""" + weights: FactorWeights = FactorWeights() + + +class QualityRequest(StrategyRequest): + """Super Quality strategy request.""" + min_fscore: int = Field(default=7, ge=0, le=9) + + +class ValueMomentumRequest(StrategyRequest): + """Value-Momentum strategy request.""" + value_weight: Decimal = Field(default=Decimal("0.5"), ge=0, le=1) + momentum_weight: Decimal = Field(default=Decimal("0.5"), ge=0, le=1) + + +class StockFactor(BaseModel): + """Factor scores for a single stock.""" + ticker: str + name: str + market: str + sector_name: Optional[str] = None + market_cap: Optional[int] = None + close_price: Optional[Decimal] = None + + # Raw metrics + per: Optional[Decimal] = None + pbr: Optional[Decimal] = None + psr: Optional[Decimal] = None + pcr: Optional[Decimal] = None + dividend_yield: Optional[Decimal] = None + roe: Optional[Decimal] = None + + # Factor scores (z-scores) + value_score: Optional[Decimal] = None + quality_score: Optional[Decimal] = None + momentum_score: Optional[Decimal] = None + + # Composite + total_score: Optional[Decimal] = None + rank: Optional[int] = None + fscore: Optional[int] = None + + +class StrategyResult(BaseModel): + """Result from running a strategy.""" + strategy_name: str + base_date: date + universe_count: int + result_count: int + stocks: List[StockFactor] + + +class StockInfo(BaseModel): + """Detailed stock information.""" + ticker: str + name: str + market: str + sector_name: Optional[str] = None + stock_type: str + close_price: Optional[Decimal] = None + market_cap: Optional[int] = None + + # Valuation + per: Optional[Decimal] = None + pbr: Optional[Decimal] = None + psr: Optional[Decimal] = None + pcr: Optional[Decimal] = None + dividend_yield: Optional[Decimal] = None + + # Per-share data + eps: Optional[Decimal] = None + bps: Optional[Decimal] = None + + base_date: Optional[date] = None + + class Config: + from_attributes = True + + +class StockSearchResult(BaseModel): + """Stock search result.""" + ticker: str + name: str + market: str + + +class PriceData(BaseModel): + """Price data point.""" + date: date + open: Decimal + high: Decimal + low: Decimal + close: Decimal + volume: int + + class Config: + from_attributes = True +``` + +**Step 2: Update schemas __init__.py** + +Add to imports and __all__: +```python +from app.schemas.strategy import ( + FactorWeights, UniverseFilter, + StrategyRequest, MultiFactorRequest, QualityRequest, ValueMomentumRequest, + StockFactor, StrategyResult, + StockInfo, StockSearchResult, PriceData, +) +``` + +**Step 3: Commit** + +```bash +git add backend/app/schemas/ +git commit -m "feat: add quant strategy Pydantic schemas" +``` + +--- + +## Task 2: Factor Calculator Service + +**Files:** +- Create: `backend/app/services/factor_calculator.py` + +**Step 1: Create factor calculator** + +```python +""" +Factor calculation service for quant strategies. +""" +from decimal import Decimal +from typing import Dict, List, Optional +from datetime import date, timedelta + +import pandas as pd +from sqlalchemy.orm import Session +from sqlalchemy import func + +from app.models.stock import Stock, Valuation, Price, Financial, Sector + + +class FactorCalculator: + """Calculates factor scores for stocks.""" + + def __init__(self, db: Session): + self.db = db + + def get_universe( + self, + markets: List[str] = None, + min_market_cap: int = None, + max_market_cap: int = None, + exclude_stock_types: List[str] = None, + exclude_sectors: List[str] = None, + ) -> List[Stock]: + """Get filtered stock universe.""" + query = self.db.query(Stock) + + if markets: + query = query.filter(Stock.market.in_(markets)) + if min_market_cap: + # market_cap is in won, min_market_cap is in 억원 + query = query.filter(Stock.market_cap >= min_market_cap * 100_000_000) + if max_market_cap: + query = query.filter(Stock.market_cap <= max_market_cap * 100_000_000) + if exclude_stock_types: + query = query.filter(~Stock.stock_type.in_(exclude_stock_types)) + + stocks = query.all() + + # Filter by sector if needed + if exclude_sectors: + sector_tickers = ( + self.db.query(Sector.ticker) + .filter(Sector.sector_name.in_(exclude_sectors)) + .all() + ) + excluded = {t[0] for t in sector_tickers} + stocks = [s for s in stocks if s.ticker not in excluded] + + return stocks + + def get_valuations(self, tickers: List[str], base_date: date = None) -> Dict[str, Valuation]: + """Get latest valuations for tickers.""" + if base_date: + valuations = ( + self.db.query(Valuation) + .filter(Valuation.ticker.in_(tickers)) + .filter(Valuation.base_date <= base_date) + .order_by(Valuation.base_date.desc()) + .all() + ) + else: + valuations = ( + self.db.query(Valuation) + .filter(Valuation.ticker.in_(tickers)) + .all() + ) + + # Get latest per ticker + result = {} + for v in valuations: + if v.ticker not in result: + result[v.ticker] = v + return result + + def get_sectors(self, tickers: List[str]) -> Dict[str, str]: + """Get sector names for tickers.""" + sectors = ( + self.db.query(Sector) + .filter(Sector.ticker.in_(tickers)) + .all() + ) + return {s.ticker: s.sector_name for s in sectors} + + def calculate_momentum( + self, + tickers: List[str], + base_date: date = None, + months: int = 12, + skip_recent: int = 1, + ) -> Dict[str, Decimal]: + """Calculate price momentum.""" + if base_date is None: + base_date = date.today() + + start_date = base_date - timedelta(days=months * 30) + skip_date = base_date - timedelta(days=skip_recent * 30) + + # Get prices + prices = ( + self.db.query(Price) + .filter(Price.ticker.in_(tickers)) + .filter(Price.date >= start_date) + .filter(Price.date <= base_date) + .all() + ) + + # Group by ticker + ticker_prices = {} + for p in prices: + if p.ticker not in ticker_prices: + ticker_prices[p.ticker] = [] + ticker_prices[p.ticker].append((p.date, float(p.close))) + + # Calculate returns + momentum = {} + for ticker, price_list in ticker_prices.items(): + if len(price_list) < 2: + continue + + price_list.sort(key=lambda x: x[0]) + + # Find start price + start_price = price_list[0][1] + + # Find end price (skip recent month if specified) + end_price = None + for d, p in reversed(price_list): + if d <= skip_date: + end_price = p + break + + if end_price and start_price > 0: + momentum[ticker] = Decimal(str((end_price - start_price) / start_price * 100)) + + return momentum + + def calculate_value_scores( + self, + valuations: Dict[str, Valuation], + ) -> Dict[str, Decimal]: + """Calculate value factor scores (higher is cheaper/better).""" + data = [] + for ticker, v in valuations.items(): + # Inverse of PER, PBR, etc. (lower ratio = higher score) + per_inv = 1 / float(v.per) if v.per and float(v.per) > 0 else 0 + pbr_inv = 1 / float(v.pbr) if v.pbr and float(v.pbr) > 0 else 0 + psr_inv = 1 / float(v.psr) if v.psr and float(v.psr) > 0 else 0 + pcr_inv = 1 / float(v.pcr) if v.pcr and float(v.pcr) > 0 else 0 + div_yield = float(v.dividend_yield) if v.dividend_yield else 0 + + data.append({ + 'ticker': ticker, + 'per_inv': per_inv, + 'pbr_inv': pbr_inv, + 'psr_inv': psr_inv, + 'pcr_inv': pcr_inv, + 'div_yield': div_yield, + }) + + if not data: + return {} + + df = pd.DataFrame(data) + + # Z-score normalization for each metric + for col in ['per_inv', 'pbr_inv', 'psr_inv', 'pcr_inv', 'div_yield']: + mean = df[col].mean() + std = df[col].std() + if std > 0: + df[f'{col}_z'] = (df[col] - mean) / std + else: + df[f'{col}_z'] = 0 + + # Composite value score (equal weight) + df['value_score'] = ( + df['per_inv_z'] + df['pbr_inv_z'] + df['psr_inv_z'] + + df['pcr_inv_z'] + df['div_yield_z'] + ) / 5 + + return {row['ticker']: Decimal(str(row['value_score'])) for _, row in df.iterrows()} + + def calculate_quality_scores( + self, + tickers: List[str], + base_date: date = None, + ) -> Dict[str, Decimal]: + """Calculate quality factor scores based on ROE, GP/A, etc.""" + # Get financial data + financials = ( + self.db.query(Financial) + .filter(Financial.ticker.in_(tickers)) + .filter(Financial.report_type == 'annual') + .all() + ) + + # Group by ticker + ticker_financials = {} + for f in financials: + if f.ticker not in ticker_financials: + ticker_financials[f.ticker] = {} + ticker_financials[f.ticker][f.account] = float(f.value) if f.value else 0 + + data = [] + for ticker, fin in ticker_financials.items(): + total_equity = fin.get('total_equity', 0) + total_assets = fin.get('total_assets', 0) + net_income = fin.get('net_income', 0) + gross_profit = fin.get('gross_profit', 0) + operating_cf = fin.get('operating_cash_flow', 0) + total_liabilities = fin.get('total_liabilities', 0) + + roe = net_income / total_equity if total_equity > 0 else 0 + gpa = gross_profit / total_assets if total_assets > 0 else 0 + cfo_a = operating_cf / total_assets if total_assets > 0 else 0 + debt_ratio_inv = 1 / (total_liabilities / total_equity) if total_equity > 0 and total_liabilities > 0 else 0 + + data.append({ + 'ticker': ticker, + 'roe': roe, + 'gpa': gpa, + 'cfo_a': cfo_a, + 'debt_ratio_inv': debt_ratio_inv, + }) + + if not data: + return {} + + df = pd.DataFrame(data) + + # Z-score normalization + for col in ['roe', 'gpa', 'cfo_a', 'debt_ratio_inv']: + mean = df[col].mean() + std = df[col].std() + if std > 0: + df[f'{col}_z'] = (df[col] - mean) / std + else: + df[f'{col}_z'] = 0 + + # Composite quality score + df['quality_score'] = (df['roe_z'] + df['gpa_z'] + df['cfo_a_z'] + df['debt_ratio_inv_z']) / 4 + + return {row['ticker']: Decimal(str(row['quality_score'])) for _, row in df.iterrows()} + + def calculate_fscore( + self, + tickers: List[str], + ) -> Dict[str, int]: + """Calculate Piotroski F-Score (0-9).""" + # Get financial data for current and previous year + financials = ( + self.db.query(Financial) + .filter(Financial.ticker.in_(tickers)) + .filter(Financial.report_type == 'annual') + .all() + ) + + # Group by ticker and date + ticker_data = {} + for f in financials: + key = (f.ticker, f.base_date) + if key not in ticker_data: + ticker_data[key] = {} + ticker_data[key][f.account] = float(f.value) if f.value else 0 + + fscores = {} + for ticker in tickers: + # Get latest two years + ticker_years = sorted( + [(k, v) for k, v in ticker_data.items() if k[0] == ticker], + key=lambda x: x[0][1], + reverse=True + )[:2] + + if len(ticker_years) < 2: + fscores[ticker] = 0 + continue + + curr = ticker_years[0][1] + prev = ticker_years[1][1] + + score = 0 + + # Profitability (4 points) + # 1. ROA > 0 + ta = curr.get('total_assets', 1) + ni = curr.get('net_income', 0) + if ta > 0 and ni / ta > 0: + score += 1 + + # 2. CFO > 0 + cfo = curr.get('operating_cash_flow', 0) + if cfo > 0: + score += 1 + + # 3. ROA increased + prev_ta = prev.get('total_assets', 1) + prev_ni = prev.get('net_income', 0) + if ta > 0 and prev_ta > 0: + if ni / ta > prev_ni / prev_ta: + score += 1 + + # 4. CFO > Net Income (Accrual) + if cfo > ni: + score += 1 + + # Leverage (3 points) + # 5. Leverage decreased + tl = curr.get('total_liabilities', 0) + prev_tl = prev.get('total_liabilities', 0) + if ta > 0 and prev_ta > 0: + if tl / ta < prev_tl / prev_ta: + score += 1 + + # 6. Liquidity increased + ca = curr.get('current_assets', 0) + cl = curr.get('current_liabilities', 1) + prev_ca = prev.get('current_assets', 0) + prev_cl = prev.get('current_liabilities', 1) + if cl > 0 and prev_cl > 0: + if ca / cl > prev_ca / prev_cl: + score += 1 + + # 7. No new equity issued (simplified: equity increase <= net income) + te = curr.get('total_equity', 0) + prev_te = prev.get('total_equity', 0) + if te - prev_te <= ni: + score += 1 + + # Operating Efficiency (2 points) + # 8. Gross margin improved + rev = curr.get('revenue', 1) + gp = curr.get('gross_profit', 0) + prev_rev = prev.get('revenue', 1) + prev_gp = prev.get('gross_profit', 0) + if rev > 0 and prev_rev > 0: + if gp / rev > prev_gp / prev_rev: + score += 1 + + # 9. Asset turnover improved + if ta > 0 and prev_ta > 0: + if rev / ta > prev_rev / prev_ta: + score += 1 + + fscores[ticker] = score + + return fscores +``` + +**Step 2: Commit** + +```bash +git add backend/app/services/factor_calculator.py +git commit -m "feat: add factor calculator service" +``` + +--- + +## Task 3: Multi-Factor Strategy Service + +**Files:** +- Create: `backend/app/services/strategy/__init__.py` +- Create: `backend/app/services/strategy/base.py` +- Create: `backend/app/services/strategy/multi_factor.py` + +**Step 1: Create strategy base class** + +```python +# backend/app/services/strategy/__init__.py +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 + +__all__ = ["BaseStrategy", "MultiFactorStrategy", "QualityStrategy", "ValueMomentumStrategy"] +``` + +```python +# backend/app/services/strategy/base.py +""" +Base strategy class. +""" +from abc import ABC, abstractmethod +from datetime import date +from typing import List + +from sqlalchemy.orm import Session + +from app.schemas.strategy import StockFactor, StrategyResult, UniverseFilter +from app.services.factor_calculator import FactorCalculator + + +class BaseStrategy(ABC): + """Base class for quant strategies.""" + + strategy_name: str = "base" + + def __init__(self, db: Session): + self.db = db + self.factor_calc = FactorCalculator(db) + + def get_universe(self, filter: UniverseFilter) -> List: + """Get filtered stock universe.""" + return self.factor_calc.get_universe( + markets=filter.markets, + min_market_cap=filter.min_market_cap, + max_market_cap=filter.max_market_cap, + exclude_stock_types=filter.exclude_stock_types, + exclude_sectors=filter.exclude_sectors, + ) + + @abstractmethod + def run( + self, + universe_filter: UniverseFilter, + top_n: int, + base_date: date = None, + **kwargs, + ) -> StrategyResult: + """Run the strategy and return ranked stocks.""" + pass +``` + +**Step 2: Create multi-factor strategy** + +```python +# backend/app/services/strategy/multi_factor.py +""" +Multi-factor strategy implementation. +""" +from datetime import date +from decimal import Decimal +from typing import List + +from sqlalchemy.orm import Session + +from app.services.strategy.base import BaseStrategy +from app.schemas.strategy import ( + StockFactor, StrategyResult, UniverseFilter, FactorWeights, +) + + +class MultiFactorStrategy(BaseStrategy): + """Multi-factor strategy combining value, quality, and momentum.""" + + strategy_name = "multi_factor" + + def run( + self, + universe_filter: UniverseFilter, + top_n: int, + base_date: date = None, + weights: FactorWeights = None, + ) -> StrategyResult: + if base_date is None: + base_date = date.today() + if weights is None: + weights = FactorWeights() + + # Get universe + stocks = self.get_universe(universe_filter) + 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 factor scores + value_scores = self.factor_calc.calculate_value_scores(valuations) + quality_scores = self.factor_calc.calculate_quality_scores(tickers, base_date) + momentum = self.factor_calc.calculate_momentum(tickers, base_date) + + # Normalize momentum to z-scores + if momentum: + mom_values = list(momentum.values()) + mom_mean = sum(mom_values) / len(mom_values) + mom_std = (sum((v - mom_mean) ** 2 for v in mom_values) / len(mom_values)) ** 0.5 + if mom_std > 0: + momentum_scores = { + t: Decimal(str((float(v) - float(mom_mean)) / float(mom_std))) + for t, v in momentum.items() + } + else: + momentum_scores = {t: Decimal("0") for t in momentum} + else: + momentum_scores = {} + + # Build result + results = [] + for ticker in tickers: + stock = stock_map[ticker] + val = valuations.get(ticker) + + v_score = value_scores.get(ticker, Decimal("0")) + q_score = quality_scores.get(ticker, Decimal("0")) + m_score = momentum_scores.get(ticker, Decimal("0")) + + # Weighted composite + total = ( + v_score * weights.value + + q_score * weights.quality + + m_score * weights.momentum + ) + + 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, + psr=Decimal(str(val.psr)) if val and val.psr else None, + pcr=Decimal(str(val.pcr)) if val and val.pcr else None, + dividend_yield=Decimal(str(val.dividend_yield)) if val and val.dividend_yield else None, + value_score=v_score, + quality_score=q_score, + momentum_score=m_score, + total_score=total, + )) + + # Sort by total score descending + results.sort(key=lambda x: x.total_score or Decimal("0"), reverse=True) + + # Assign ranks and limit + 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], + ) +``` + +**Step 3: Commit** + +```bash +git add backend/app/services/strategy/ +git commit -m "feat: add multi-factor strategy service" +``` + +--- + +## Task 4: Quality and Value-Momentum Strategies + +**Files:** +- Create: `backend/app/services/strategy/quality.py` +- Create: `backend/app/services/strategy/value_momentum.py` + +**Step 1: Create quality strategy** + +```python +# backend/app/services/strategy/quality.py +""" +Super Quality strategy implementation. +""" +from datetime import date +from decimal import Decimal + +from app.services.strategy.base import BaseStrategy +from app.schemas.strategy import StockFactor, StrategyResult, UniverseFilter + + +class QualityStrategy(BaseStrategy): + """Super Quality strategy using F-Score and quality factors.""" + + strategy_name = "quality" + + def run( + self, + universe_filter: UniverseFilter, + top_n: int, + base_date: date = None, + min_fscore: int = 7, + ) -> StrategyResult: + if base_date is None: + base_date = date.today() + + # Get universe + stocks = self.get_universe(universe_filter) + 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 data + valuations = self.factor_calc.get_valuations(tickers, base_date) + sectors = self.factor_calc.get_sectors(tickers) + fscores = self.factor_calc.calculate_fscore(tickers) + quality_scores = self.factor_calc.calculate_quality_scores(tickers, base_date) + + # Filter by F-Score + qualified_tickers = [t for t in tickers if fscores.get(t, 0) >= min_fscore] + + # Build results + results = [] + for ticker in qualified_tickers: + stock = stock_map[ticker] + val = valuations.get(ticker) + q_score = quality_scores.get(ticker, Decimal("0")) + fscore = fscores.get(ticker, 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, + quality_score=q_score, + total_score=q_score, + fscore=fscore, + )) + + # Sort by quality score + 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], + ) +``` + +**Step 2: Create value-momentum strategy** + +```python +# backend/app/services/strategy/value_momentum.py +""" +Value-Momentum strategy implementation. +""" +from datetime import date +from decimal import Decimal + +from app.services.strategy.base import BaseStrategy +from app.schemas.strategy import StockFactor, StrategyResult, UniverseFilter + + +class ValueMomentumStrategy(BaseStrategy): + """Value-Momentum combined strategy.""" + + strategy_name = "value_momentum" + + def run( + self, + universe_filter: UniverseFilter, + top_n: int, + base_date: date = None, + value_weight: Decimal = Decimal("0.5"), + momentum_weight: Decimal = Decimal("0.5"), + ) -> StrategyResult: + if base_date is None: + base_date = date.today() + + # Get universe + stocks = self.get_universe(universe_filter) + 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 data + valuations = self.factor_calc.get_valuations(tickers, base_date) + sectors = self.factor_calc.get_sectors(tickers) + value_scores = self.factor_calc.calculate_value_scores(valuations) + momentum = self.factor_calc.calculate_momentum(tickers, base_date) + + # Normalize momentum + if momentum: + mom_values = list(momentum.values()) + mom_mean = sum(mom_values) / len(mom_values) + mom_std = (sum((v - mom_mean) ** 2 for v in mom_values) / len(mom_values)) ** 0.5 + if mom_std > 0: + momentum_scores = { + t: Decimal(str((float(v) - float(mom_mean)) / float(mom_std))) + for t, v in momentum.items() + } + else: + momentum_scores = {t: Decimal("0") for t in momentum} + else: + momentum_scores = {} + + # Build results + results = [] + for ticker in tickers: + stock = stock_map[ticker] + val = valuations.get(ticker) + v_score = value_scores.get(ticker, Decimal("0")) + m_score = momentum_scores.get(ticker, Decimal("0")) + + total = v_score * value_weight + m_score * momentum_weight + + 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, + value_score=v_score, + momentum_score=m_score, + total_score=total, + )) + + 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], + ) +``` + +**Step 3: Update __init__.py to include new strategies** + +Already done in Task 3. + +**Step 4: Commit** + +```bash +git add backend/app/services/strategy/ +git commit -m "feat: add quality and value-momentum strategies" +``` + +--- + +## Task 5: Strategy API Endpoints + +**Files:** +- Create: `backend/app/api/strategy.py` +- Update: `backend/app/api/__init__.py` +- Update: `backend/app/main.py` + +**Step 1: Create strategy API** + +```python +# backend/app/api/strategy.py +""" +Quant strategy API endpoints. +""" +from fastapi import APIRouter, Depends +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, +) +from app.services.strategy import MultiFactorStrategy, QualityStrategy, ValueMomentumStrategy + +router = APIRouter(prefix="/api/strategy", tags=["strategy"]) + + +@router.post("/multi-factor", response_model=StrategyResult) +async def run_multi_factor( + request: MultiFactorRequest, + current_user: CurrentUser, + db: Session = Depends(get_db), +): + """Run multi-factor strategy.""" + strategy = MultiFactorStrategy(db) + return strategy.run( + universe_filter=request.universe, + top_n=request.top_n, + base_date=request.base_date, + weights=request.weights, + ) + + +@router.post("/quality", response_model=StrategyResult) +async def run_quality( + request: QualityRequest, + current_user: CurrentUser, + db: Session = Depends(get_db), +): + """Run super quality strategy.""" + strategy = QualityStrategy(db) + return strategy.run( + universe_filter=request.universe, + top_n=request.top_n, + base_date=request.base_date, + min_fscore=request.min_fscore, + ) + + +@router.post("/value-momentum", response_model=StrategyResult) +async def run_value_momentum( + request: ValueMomentumRequest, + current_user: CurrentUser, + db: Session = Depends(get_db), +): + """Run value-momentum strategy.""" + strategy = ValueMomentumStrategy(db) + return strategy.run( + universe_filter=request.universe, + top_n=request.top_n, + base_date=request.base_date, + value_weight=request.value_weight, + momentum_weight=request.momentum_weight, + ) +``` + +**Step 2: Update api/__init__.py** + +Add: +```python +from app.api.strategy import router as strategy_router + +__all__ = ["auth_router", "admin_router", "portfolio_router", "strategy_router"] +``` + +**Step 3: Update main.py** + +Add: +```python +from app.api import auth_router, admin_router, portfolio_router, strategy_router + +app.include_router(strategy_router) +``` + +**Step 4: Commit** + +```bash +git add backend/app/api/ +git add backend/app/main.py +git commit -m "feat: add strategy API endpoints" +``` + +--- + +## Task 6: Market API Endpoints + +**Files:** +- Create: `backend/app/api/market.py` +- Update: `backend/app/api/__init__.py` +- Update: `backend/app/main.py` + +**Step 1: Create market API** + +```python +# backend/app/api/market.py +""" +Market data API endpoints. +""" +from typing import List, Optional +from datetime import date, timedelta + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.orm import Session + +from app.core.database import get_db +from app.api.deps import CurrentUser +from app.models.stock import Stock, Valuation, Price, Sector +from app.schemas.strategy import StockInfo, StockSearchResult, PriceData + +router = APIRouter(prefix="/api/market", tags=["market"]) + + +@router.get("/stocks/{ticker}", response_model=StockInfo) +async def get_stock( + ticker: str, + current_user: CurrentUser, + db: Session = Depends(get_db), +): + """Get stock information.""" + stock = db.query(Stock).filter(Stock.ticker == ticker).first() + if not stock: + raise HTTPException(status_code=404, detail="Stock not found") + + valuation = ( + db.query(Valuation) + .filter(Valuation.ticker == ticker) + .order_by(Valuation.base_date.desc()) + .first() + ) + + sector = db.query(Sector).filter(Sector.ticker == ticker).first() + + return StockInfo( + ticker=stock.ticker, + name=stock.name, + market=stock.market, + sector_name=sector.sector_name if sector else None, + stock_type=stock.stock_type.value if stock.stock_type else "common", + close_price=stock.close_price, + market_cap=int(stock.market_cap / 100_000_000) if stock.market_cap else None, + per=valuation.per if valuation else None, + pbr=valuation.pbr if valuation else None, + psr=valuation.psr if valuation else None, + pcr=valuation.pcr if valuation else None, + dividend_yield=valuation.dividend_yield if valuation else None, + eps=stock.eps, + bps=stock.bps, + base_date=stock.base_date, + ) + + +@router.get("/stocks/{ticker}/prices", response_model=List[PriceData]) +async def get_stock_prices( + ticker: str, + current_user: CurrentUser, + db: Session = Depends(get_db), + days: int = Query(default=365, ge=1, le=3650), +): + """Get historical prices for a stock.""" + start_date = date.today() - timedelta(days=days) + + prices = ( + db.query(Price) + .filter(Price.ticker == ticker) + .filter(Price.date >= start_date) + .order_by(Price.date.desc()) + .all() + ) + + return [ + PriceData( + date=p.date, + open=p.open, + high=p.high, + low=p.low, + close=p.close, + volume=p.volume, + ) + for p in prices + ] + + +@router.get("/search", response_model=List[StockSearchResult]) +async def search_stocks( + q: str = Query(..., min_length=1), + current_user: CurrentUser = None, + db: Session = Depends(get_db), + limit: int = Query(default=20, ge=1, le=100), +): + """Search stocks by ticker or name.""" + stocks = ( + db.query(Stock) + .filter( + (Stock.ticker.ilike(f"%{q}%")) | (Stock.name.ilike(f"%{q}%")) + ) + .limit(limit) + .all() + ) + + return [ + StockSearchResult( + ticker=s.ticker, + name=s.name, + market=s.market, + ) + for s in stocks + ] +``` + +**Step 2: Update api/__init__.py** + +Add: +```python +from app.api.market import router as market_router + +__all__ = ["auth_router", "admin_router", "portfolio_router", "strategy_router", "market_router"] +``` + +**Step 3: Update main.py** + +Add to imports and include router. + +**Step 4: Commit** + +```bash +git add backend/app/api/ +git add backend/app/main.py +git commit -m "feat: add market data API endpoints" +``` + +--- + +## Task 7: Frontend Strategy List Page + +**Files:** +- Create: `frontend/src/app/strategy/page.tsx` + +**Step 1: Create strategy list page** + +```typescript +'use client'; + +import { useEffect, useState } from 'react'; +import { useRouter } from 'next/navigation'; +import Link from 'next/link'; +import Sidebar from '@/components/layout/Sidebar'; +import Header from '@/components/layout/Header'; +import { api } from '@/lib/api'; + +interface User { + id: number; + username: string; +} + +const strategies = [ + { + id: 'multi-factor', + name: '멀티 팩터', + description: '밸류, 퀄리티, 모멘텀 팩터를 조합한 종합 전략', + icon: '📊', + }, + { + id: 'quality', + name: '슈퍼 퀄리티', + description: 'F-Score 기반 우량주 선별 전략', + icon: '⭐', + }, + { + id: 'value-momentum', + name: '밸류 모멘텀', + description: '가치주와 모멘텀을 결합한 전략', + icon: '📈', + }, +]; + +export default function StrategyListPage() { + const router = useRouter(); + const [user, setUser] = useState(null); + const [loading, setLoading] = useState(true); + + useEffect(() => { + const init = async () => { + try { + const userData = await api.getCurrentUser() as User; + setUser(userData); + } catch { + router.push('/login'); + } finally { + setLoading(false); + } + }; + init(); + }, [router]); + + if (loading) { + return ( +
+
Loading...
+
+ ); + } + + return ( +
+ +
+
+
+

퀀트 전략

+ +
+ {strategies.map((strategy) => ( + +
{strategy.icon}
+

+ {strategy.name} +

+

{strategy.description}

+ + ))} +
+
+
+
+ ); +} +``` + +**Step 2: Commit** + +```bash +git add frontend/src/app/strategy/ +git commit -m "feat: add strategy list page" +``` + +--- + +## Task 8: Frontend Multi-Factor Strategy Page + +**Files:** +- Create: `frontend/src/app/strategy/multi-factor/page.tsx` + +**Step 1: Create multi-factor strategy page** + +```typescript +'use client'; + +import { useState } from 'react'; +import { useRouter } from 'next/navigation'; +import Sidebar from '@/components/layout/Sidebar'; +import Header from '@/components/layout/Header'; +import { api } from '@/lib/api'; + +interface StockFactor { + ticker: string; + name: string; + market: string; + sector_name: string | null; + market_cap: number | null; + close_price: number | null; + per: number | null; + pbr: number | null; + value_score: number | null; + quality_score: number | null; + momentum_score: number | null; + total_score: number | null; + rank: number | null; +} + +interface StrategyResult { + strategy_name: string; + base_date: string; + universe_count: number; + result_count: number; + stocks: StockFactor[]; +} + +export default function MultiFactorPage() { + const router = useRouter(); + const [loading, setLoading] = useState(false); + const [result, setResult] = useState(null); + const [error, setError] = useState(null); + + // Weights + const [valueWeight, setValueWeight] = useState(0.25); + const [qualityWeight, setQualityWeight] = useState(0.25); + const [momentumWeight, setMomentumWeight] = useState(0.25); + const [topN, setTopN] = useState(30); + + const runStrategy = async () => { + setLoading(true); + setError(null); + try { + const data = await api.post('/api/strategy/multi-factor', { + universe: { + markets: ['KOSPI', 'KOSDAQ'], + exclude_stock_types: ['spac', 'preferred', 'reit'], + }, + top_n: topN, + weights: { + value: valueWeight, + quality: qualityWeight, + momentum: momentumWeight, + low_vol: 1 - valueWeight - qualityWeight - momentumWeight, + }, + }); + setResult(data); + } catch (err) { + setError(err instanceof Error ? err.message : 'Strategy execution failed'); + } finally { + setLoading(false); + } + }; + + const formatNumber = (value: number | null, decimals: number = 2) => { + if (value === null) return '-'; + return value.toFixed(decimals); + }; + + const formatCurrency = (value: number | null) => { + if (value === null) return '-'; + return new Intl.NumberFormat('ko-KR').format(value); + }; + + return ( +
+ +
+
+
+

멀티 팩터 전략

+ + {error && ( +
+ {error} +
+ )} + + {/* Settings */} +
+

전략 설정

+
+
+ + setValueWeight(parseFloat(e.target.value))} + className="w-full px-3 py-2 border rounded" + /> +
+
+ + setQualityWeight(parseFloat(e.target.value))} + className="w-full px-3 py-2 border rounded" + /> +
+
+ + setMomentumWeight(parseFloat(e.target.value))} + className="w-full px-3 py-2 border rounded" + /> +
+
+ + setTopN(parseInt(e.target.value))} + className="w-full px-3 py-2 border rounded" + /> +
+
+ +
+ + {/* Results */} + {result && ( +
+
+

+ 결과 ({result.result_count}/{result.universe_count} 종목) +

+

기준일: {result.base_date}

+
+
+ + + + + + + + + + + + + + + + + + {result.stocks.map((stock) => ( + + + + + + + + + + + + + + ))} + +
순위종목섹터시가총액(억)현재가PERPBR밸류퀄리티모멘텀종합
{stock.rank} +
{stock.ticker}
+
{stock.name}
+ 		{stock.sector_name || '-'}{formatCurrency(stock.market_cap)}{formatCurrency(stock.close_price)}{formatNumber(stock.per)}{formatNumber(stock.pbr)}{formatNumber(stock.value_score)}{formatNumber(stock.quality_score)}{formatNumber(stock.momentum_score)}{formatNumber(stock.total_score)}
+
+
+ )} +
+
+
+ ); +} +``` + +**Step 2: Commit** + +```bash +git add frontend/src/app/strategy/multi-factor/ +git commit -m "feat: add multi-factor strategy page" +``` + +--- + +## Task 9: Frontend Quality Strategy Page + +**Files:** +- Create: `frontend/src/app/strategy/quality/page.tsx` + +**Step 1: Create quality strategy page** + +```typescript +'use client'; + +import { useState } from 'react'; +import Sidebar from '@/components/layout/Sidebar'; +import Header from '@/components/layout/Header'; +import { api } from '@/lib/api'; + +interface StockFactor { + ticker: string; + name: string; + market: string; + sector_name: string | null; + market_cap: number | null; + close_price: number | null; + per: number | null; + pbr: number | null; + dividend_yield: number | null; + quality_score: number | null; + fscore: number | null; + rank: number | null; +} + +interface StrategyResult { + strategy_name: string; + base_date: string; + universe_count: number; + result_count: number; + stocks: StockFactor[]; +} + +export default function QualityStrategyPage() { + const [loading, setLoading] = useState(false); + const [result, setResult] = useState(null); + const [error, setError] = useState(null); + + const [minFscore, setMinFscore] = useState(7); + const [topN, setTopN] = useState(30); + + const runStrategy = async () => { + setLoading(true); + setError(null); + try { + const data = await api.post('/api/strategy/quality', { + universe: { + markets: ['KOSPI', 'KOSDAQ'], + exclude_stock_types: ['spac', 'preferred', 'reit'], + }, + top_n: topN, + min_fscore: minFscore, + }); + setResult(data); + } catch (err) { + setError(err instanceof Error ? err.message : 'Strategy execution failed'); + } finally { + setLoading(false); + } + }; + + const formatNumber = (value: number | null, decimals: number = 2) => { + if (value === null) return '-'; + return value.toFixed(decimals); + }; + + const formatCurrency = (value: number | null) => { + if (value === null) return '-'; + return new Intl.NumberFormat('ko-KR').format(value); + }; + + return ( +
+ +
+
+
+

슈퍼 퀄리티 전략

+ + {error && ( +
+ {error} +
+ )} + + {/* Settings */} +
+

전략 설정

+
+
+ + setMinFscore(parseInt(e.target.value))} + className="w-full px-3 py-2 border rounded" + /> +
+
+ + setTopN(parseInt(e.target.value))} + className="w-full px-3 py-2 border rounded" + /> +
+
+ +
+ + {/* Results */} + {result && ( +
+
+

+ 결과 ({result.result_count}/{result.universe_count} 종목) +

+

기준일: {result.base_date}

+
+
+ + + + + + + + + + + + + + + + + {result.stocks.map((stock) => ( + + + + + + + + + + + + + ))} + +
순위종목섹터시가총액(억)현재가PERPBR배당률F-Score퀄리티
{stock.rank} +
{stock.ticker}
+
{stock.name}
+ 		{stock.sector_name || '-'}{formatCurrency(stock.market_cap)}{formatCurrency(stock.close_price)}{formatNumber(stock.per)}{formatNumber(stock.pbr)}{formatNumber(stock.dividend_yield)}% + = 8 ? 'bg-green-100 text-green-800' : + (stock.fscore ?? 0) >= 6 ? 'bg-yellow-100 text-yellow-800' : + 'bg-gray-100 text-gray-800' + }`}> + {stock.fscore}/9 + + {formatNumber(stock.quality_score)}
+
+
+ )} +
+
+
+ ); +} +``` + +**Step 2: Commit** + +```bash +git add frontend/src/app/strategy/quality/ +git commit -m "feat: add quality strategy page" +``` + +--- + +## Task 10: Frontend Value-Momentum Strategy Page + +**Files:** +- Create: `frontend/src/app/strategy/value-momentum/page.tsx` + +**Step 1: Create value-momentum strategy page** + +```typescript +'use client'; + +import { useState } from 'react'; +import Sidebar from '@/components/layout/Sidebar'; +import Header from '@/components/layout/Header'; +import { api } from '@/lib/api'; + +interface StockFactor { + ticker: string; + name: string; + market: string; + sector_name: string | null; + market_cap: number | null; + close_price: number | null; + per: number | null; + pbr: number | null; + dividend_yield: number | null; + value_score: number | null; + momentum_score: number | null; + total_score: number | null; + rank: number | null; +} + +interface StrategyResult { + strategy_name: string; + base_date: string; + universe_count: number; + result_count: number; + stocks: StockFactor[]; +} + +export default function ValueMomentumPage() { + const [loading, setLoading] = useState(false); + const [result, setResult] = useState(null); + const [error, setError] = useState(null); + + const [valueWeight, setValueWeight] = useState(0.5); + const [momentumWeight, setMomentumWeight] = useState(0.5); + const [topN, setTopN] = useState(30); + + const runStrategy = async () => { + setLoading(true); + setError(null); + try { + const data = await api.post('/api/strategy/value-momentum', { + universe: { + markets: ['KOSPI', 'KOSDAQ'], + exclude_stock_types: ['spac', 'preferred', 'reit'], + }, + top_n: topN, + value_weight: valueWeight, + momentum_weight: momentumWeight, + }); + setResult(data); + } catch (err) { + setError(err instanceof Error ? err.message : 'Strategy execution failed'); + } finally { + setLoading(false); + } + }; + + const formatNumber = (value: number | null, decimals: number = 2) => { + if (value === null) return '-'; + return value.toFixed(decimals); + }; + + const formatCurrency = (value: number | null) => { + if (value === null) return '-'; + return new Intl.NumberFormat('ko-KR').format(value); + }; + + return ( +
+ +
+
+
+

밸류 모멘텀 전략

+ + {error && ( +
+ {error} +
+ )} + + {/* Settings */} +
+

전략 설정

+
+
+ + setValueWeight(parseFloat(e.target.value))} + className="w-full px-3 py-2 border rounded" + /> +
+
+ + setMomentumWeight(parseFloat(e.target.value))} + className="w-full px-3 py-2 border rounded" + /> +
+
+ + setTopN(parseInt(e.target.value))} + className="w-full px-3 py-2 border rounded" + /> +
+
+ +
+ + {/* Results */} + {result && ( +
+
+

+ 결과 ({result.result_count}/{result.universe_count} 종목) +

+

기준일: {result.base_date}

+
+
+ + + + + + + + + + + + + + + + + {result.stocks.map((stock) => ( + + + + + + + + + + + + + ))} + +
순위종목섹터시가총액(억)현재가PERPBR밸류모멘텀종합
{stock.rank} +
{stock.ticker}
+
{stock.name}
+ 		{stock.sector_name || '-'}{formatCurrency(stock.market_cap)}{formatCurrency(stock.close_price)}{formatNumber(stock.per)}{formatNumber(stock.pbr)}{formatNumber(stock.value_score)}{formatNumber(stock.momentum_score)}{formatNumber(stock.total_score)}
+
+
+ )} +
+
+
+ ); +} +``` + +**Step 2: Commit** + +```bash +git add frontend/src/app/strategy/value-momentum/ +git commit -m "feat: add value-momentum strategy page" +``` + +--- + +## Task 11: Verify Phase 4 Integration + +**Step 1: Verify all backend files exist** + +Check: +- backend/app/schemas/strategy.py +- backend/app/services/factor_calculator.py +- backend/app/services/strategy/base.py +- backend/app/services/strategy/multi_factor.py +- backend/app/services/strategy/quality.py +- backend/app/services/strategy/value_momentum.py +- backend/app/api/strategy.py +- backend/app/api/market.py + +**Step 2: Verify all frontend files exist** + +Check: +- frontend/src/app/strategy/page.tsx +- frontend/src/app/strategy/multi-factor/page.tsx +- frontend/src/app/strategy/quality/page.tsx +- frontend/src/app/strategy/value-momentum/page.tsx + +**Step 3: Run frontend build** + +```bash +cd /home/zephyrdark/workspace/quant/galaxy-po/frontend +npm run build +``` + +**Step 4: Show git log** + +```bash +cd /home/zephyrdark/workspace/quant/galaxy-po +git log --oneline -15 +``` + +--- + +## Summary + +Phase 4 완료 시 구현된 기능: +- Strategy Pydantic 스키마 (FactorWeights, UniverseFilter, StrategyRequest, StockFactor, etc.) +- Factor Calculator 서비스 (밸류, 퀄리티, 모멘텀, F-Score 계산) +- Multi-Factor 전략 (가중치 기반 종합 점수) +- Super Quality 전략 (F-Score 필터 + 퀄리티 점수) +- Value-Momentum 전략 (밸류 + 모멘텀 조합) +- Strategy API 엔드포인트 (multi-factor, quality, value-momentum) +- Market API 엔드포인트 (종목 조회, 시세 조회, 검색) +- Frontend 전략 목록 페이지 +- Frontend 멀티 팩터 전략 페이지 +- Frontend 슈퍼 퀄리티 전략 페이지 +- Frontend 밸류 모멘텀 전략 페이지 + +다음 Phase: 백테스트 엔진 구현