f0ae25c533
Some checks failed
CI / test (pull_request) Has been cancelled
Add Python-first scanning pipeline that reduces Gemini API calls by filtering stocks before AI analysis: KIS rankings API -> RSI/volume filter -> AI judgment. ## Implementation - Add RSI calculation (Wilder's smoothing method) to VolatilityAnalyzer - Add KIS API methods: fetch_market_rankings() and get_daily_prices() - Create SmartVolatilityScanner with configurable thresholds - Integrate scanner into main.py realtime mode - Add selection_context logging to trades table for Evolution system ## Configuration - RSI_OVERSOLD_THRESHOLD: 30 (configurable 0-50) - RSI_MOMENTUM_THRESHOLD: 70 (configurable 50-100) - VOL_MULTIPLIER: 2.0 (minimum volume ratio, configurable 1-10) - SCANNER_TOP_N: 3 (max candidates per scan, configurable 1-10) ## Benefits - Reduces Gemini API calls (process 1-3 qualified stocks vs 20-30 ranked) - Python-based technical filtering before expensive AI judgment - Tracks selection criteria (RSI, volume_ratio, signal, score) for strategy optimization - Graceful fallback to static watchlist if ranking API fails ## Tests - 13 new tests for SmartVolatilityScanner and RSI calculation - All existing tests updated and passing - Coverage maintained at 73% Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
374 lines
12 KiB
Python
374 lines
12 KiB
Python
"""Volatility and momentum analysis for stock selection.
|
|
|
|
Calculates ATR, price change percentages, volume surges, and price-volume divergence.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class VolatilityMetrics:
|
|
"""Volatility and momentum metrics for a stock."""
|
|
|
|
stock_code: str
|
|
current_price: float
|
|
atr: float # Average True Range (14 periods)
|
|
price_change_1m: float # 1-minute price change %
|
|
price_change_5m: float # 5-minute price change %
|
|
price_change_15m: float # 15-minute price change %
|
|
volume_surge: float # Volume vs average (ratio)
|
|
pv_divergence: float # Price-volume divergence score
|
|
momentum_score: float # Combined momentum score (0-100)
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"VolatilityMetrics({self.stock_code}: "
|
|
f"price={self.current_price:.2f}, "
|
|
f"atr={self.atr:.2f}, "
|
|
f"1m={self.price_change_1m:.2f}%, "
|
|
f"vol_surge={self.volume_surge:.2f}x, "
|
|
f"momentum={self.momentum_score:.1f})"
|
|
)
|
|
|
|
|
|
class VolatilityAnalyzer:
|
|
"""Analyzes stock volatility and momentum for leader detection."""
|
|
|
|
def __init__(self, min_volume_surge: float = 2.0, min_price_change: float = 1.0) -> None:
|
|
"""Initialize the volatility analyzer.
|
|
|
|
Args:
|
|
min_volume_surge: Minimum volume surge ratio (default 2x average)
|
|
min_price_change: Minimum price change % for breakout (default 1%)
|
|
"""
|
|
self.min_volume_surge = min_volume_surge
|
|
self.min_price_change = min_price_change
|
|
|
|
def calculate_atr(
|
|
self,
|
|
high_prices: list[float],
|
|
low_prices: list[float],
|
|
close_prices: list[float],
|
|
period: int = 14,
|
|
) -> float:
|
|
"""Calculate Average True Range (ATR).
|
|
|
|
Args:
|
|
high_prices: List of high prices (most recent last)
|
|
low_prices: List of low prices (most recent last)
|
|
close_prices: List of close prices (most recent last)
|
|
period: ATR period (default 14)
|
|
|
|
Returns:
|
|
ATR value
|
|
"""
|
|
if (
|
|
len(high_prices) < period + 1
|
|
or len(low_prices) < period + 1
|
|
or len(close_prices) < period + 1
|
|
):
|
|
return 0.0
|
|
|
|
true_ranges: list[float] = []
|
|
for i in range(1, len(high_prices)):
|
|
high = high_prices[i]
|
|
low = low_prices[i]
|
|
prev_close = close_prices[i - 1]
|
|
|
|
tr = max(
|
|
high - low,
|
|
abs(high - prev_close),
|
|
abs(low - prev_close),
|
|
)
|
|
true_ranges.append(tr)
|
|
|
|
if len(true_ranges) < period:
|
|
return 0.0
|
|
|
|
# Simple Moving Average of True Range
|
|
recent_tr = true_ranges[-period:]
|
|
return sum(recent_tr) / len(recent_tr)
|
|
|
|
def calculate_price_change(
|
|
self, current_price: float, past_price: float
|
|
) -> float:
|
|
"""Calculate price change percentage.
|
|
|
|
Args:
|
|
current_price: Current price
|
|
past_price: Past price to compare against
|
|
|
|
Returns:
|
|
Price change percentage
|
|
"""
|
|
if past_price == 0:
|
|
return 0.0
|
|
return ((current_price - past_price) / past_price) * 100
|
|
|
|
def calculate_volume_surge(
|
|
self, current_volume: float, avg_volume: float
|
|
) -> float:
|
|
"""Calculate volume surge ratio.
|
|
|
|
Args:
|
|
current_volume: Current volume
|
|
avg_volume: Average volume
|
|
|
|
Returns:
|
|
Volume surge ratio (current / average)
|
|
"""
|
|
if avg_volume == 0:
|
|
return 1.0
|
|
return current_volume / avg_volume
|
|
|
|
def calculate_rsi(
|
|
self,
|
|
close_prices: list[float],
|
|
period: int = 14,
|
|
) -> float:
|
|
"""Calculate Relative Strength Index (RSI) using Wilder's smoothing.
|
|
|
|
Args:
|
|
close_prices: List of closing prices (oldest to newest, minimum period+1 values)
|
|
period: RSI period (default 14)
|
|
|
|
Returns:
|
|
RSI value between 0 and 100, or 50.0 (neutral) if insufficient data
|
|
|
|
Examples:
|
|
>>> analyzer = VolatilityAnalyzer()
|
|
>>> prices = [100 - i * 0.5 for i in range(20)] # Downtrend
|
|
>>> rsi = analyzer.calculate_rsi(prices)
|
|
>>> assert rsi < 50 # Oversold territory
|
|
"""
|
|
if len(close_prices) < period + 1:
|
|
return 50.0 # Neutral RSI if insufficient data
|
|
|
|
# Calculate price changes
|
|
changes = [close_prices[i] - close_prices[i - 1] for i in range(1, len(close_prices))]
|
|
|
|
# Separate gains and losses
|
|
gains = [max(0.0, change) for change in changes]
|
|
losses = [max(0.0, -change) for change in changes]
|
|
|
|
# Calculate initial average gain/loss (simple average for first period)
|
|
avg_gain = sum(gains[:period]) / period
|
|
avg_loss = sum(losses[:period]) / period
|
|
|
|
# Apply Wilder's smoothing for remaining periods
|
|
for i in range(period, len(changes)):
|
|
avg_gain = (avg_gain * (period - 1) + gains[i]) / period
|
|
avg_loss = (avg_loss * (period - 1) + losses[i]) / period
|
|
|
|
# Calculate RS and RSI
|
|
if avg_loss == 0:
|
|
return 100.0 # All gains, maximum RSI
|
|
|
|
rs = avg_gain / avg_loss
|
|
rsi = 100 - (100 / (1 + rs))
|
|
|
|
return rsi
|
|
|
|
def calculate_pv_divergence(
|
|
self,
|
|
price_change: float,
|
|
volume_surge: float,
|
|
) -> float:
|
|
"""Calculate price-volume divergence score.
|
|
|
|
Positive divergence: Price up + Volume up = bullish
|
|
Negative divergence: Price up + Volume down = bearish
|
|
Neutral: Price/volume move together moderately
|
|
|
|
Args:
|
|
price_change: Price change percentage
|
|
volume_surge: Volume surge ratio
|
|
|
|
Returns:
|
|
Divergence score (-100 to +100)
|
|
"""
|
|
# Normalize volume surge to -1 to +1 scale (1.0 = neutral)
|
|
volume_signal = (volume_surge - 1.0) * 10 # Scale for sensitivity
|
|
|
|
# Calculate divergence
|
|
# Positive: price and volume move in same direction
|
|
# Negative: price and volume move in opposite directions
|
|
if price_change > 0 and volume_surge > 1.0:
|
|
# Bullish: price up, volume up
|
|
return min(100.0, price_change * volume_signal)
|
|
elif price_change < 0 and volume_surge < 1.0:
|
|
# Bearish confirmation: price down, volume down
|
|
return max(-100.0, price_change * volume_signal)
|
|
elif price_change > 0 and volume_surge < 1.0:
|
|
# Bearish divergence: price up but volume low (weak rally)
|
|
return -abs(price_change) * 0.5
|
|
elif price_change < 0 and volume_surge > 1.0:
|
|
# Selling pressure: price down, volume up
|
|
return price_change * volume_signal
|
|
else:
|
|
return 0.0
|
|
|
|
def calculate_momentum_score(
|
|
self,
|
|
price_change_1m: float,
|
|
price_change_5m: float,
|
|
price_change_15m: float,
|
|
volume_surge: float,
|
|
atr: float,
|
|
current_price: float,
|
|
) -> float:
|
|
"""Calculate combined momentum score (0-100).
|
|
|
|
Weights:
|
|
- 1m change: 40%
|
|
- 5m change: 30%
|
|
- 15m change: 20%
|
|
- Volume surge: 10%
|
|
|
|
Args:
|
|
price_change_1m: 1-minute price change %
|
|
price_change_5m: 5-minute price change %
|
|
price_change_15m: 15-minute price change %
|
|
volume_surge: Volume surge ratio
|
|
atr: Average True Range
|
|
current_price: Current price
|
|
|
|
Returns:
|
|
Momentum score (0-100)
|
|
"""
|
|
# Weight recent changes more heavily
|
|
weighted_change = (
|
|
price_change_1m * 0.4 +
|
|
price_change_5m * 0.3 +
|
|
price_change_15m * 0.2
|
|
)
|
|
|
|
# Volume contribution (normalized to 0-10 scale)
|
|
volume_contribution = min(10.0, (volume_surge - 1.0) * 5.0)
|
|
|
|
# Volatility bonus: higher ATR = higher potential (normalized)
|
|
volatility_bonus = 0.0
|
|
if current_price > 0:
|
|
atr_pct = (atr / current_price) * 100
|
|
volatility_bonus = min(10.0, atr_pct)
|
|
|
|
# Combine scores
|
|
raw_score = weighted_change + volume_contribution + volatility_bonus
|
|
|
|
# Normalize to 0-100 scale
|
|
# Assume typical momentum range is -10 to +30
|
|
normalized = ((raw_score + 10) / 40) * 100
|
|
|
|
return max(0.0, min(100.0, normalized))
|
|
|
|
def analyze(
|
|
self,
|
|
stock_code: str,
|
|
orderbook_data: dict[str, Any],
|
|
price_history: dict[str, Any],
|
|
) -> VolatilityMetrics:
|
|
"""Analyze volatility and momentum for a stock.
|
|
|
|
Args:
|
|
stock_code: Stock code
|
|
orderbook_data: Current orderbook/quote data
|
|
price_history: Historical price and volume data
|
|
|
|
Returns:
|
|
VolatilityMetrics with calculated indicators
|
|
"""
|
|
# Extract current data from orderbook
|
|
output1 = orderbook_data.get("output1", {})
|
|
current_price = float(output1.get("stck_prpr", 0))
|
|
current_volume = float(output1.get("acml_vol", 0))
|
|
|
|
# Extract historical data
|
|
high_prices = price_history.get("high", [])
|
|
low_prices = price_history.get("low", [])
|
|
close_prices = price_history.get("close", [])
|
|
volumes = price_history.get("volume", [])
|
|
|
|
# Calculate ATR
|
|
atr = self.calculate_atr(high_prices, low_prices, close_prices)
|
|
|
|
# Calculate price changes (use historical data if available)
|
|
price_change_1m = 0.0
|
|
price_change_5m = 0.0
|
|
price_change_15m = 0.0
|
|
|
|
if len(close_prices) > 0:
|
|
if len(close_prices) >= 1:
|
|
price_change_1m = self.calculate_price_change(
|
|
current_price, close_prices[-1]
|
|
)
|
|
if len(close_prices) >= 5:
|
|
price_change_5m = self.calculate_price_change(
|
|
current_price, close_prices[-5]
|
|
)
|
|
if len(close_prices) >= 15:
|
|
price_change_15m = self.calculate_price_change(
|
|
current_price, close_prices[-15]
|
|
)
|
|
|
|
# Calculate volume surge
|
|
avg_volume = sum(volumes) / len(volumes) if volumes else current_volume
|
|
volume_surge = self.calculate_volume_surge(current_volume, avg_volume)
|
|
|
|
# Calculate price-volume divergence
|
|
pv_divergence = self.calculate_pv_divergence(price_change_1m, volume_surge)
|
|
|
|
# Calculate momentum score
|
|
momentum_score = self.calculate_momentum_score(
|
|
price_change_1m,
|
|
price_change_5m,
|
|
price_change_15m,
|
|
volume_surge,
|
|
atr,
|
|
current_price,
|
|
)
|
|
|
|
return VolatilityMetrics(
|
|
stock_code=stock_code,
|
|
current_price=current_price,
|
|
atr=atr,
|
|
price_change_1m=price_change_1m,
|
|
price_change_5m=price_change_5m,
|
|
price_change_15m=price_change_15m,
|
|
volume_surge=volume_surge,
|
|
pv_divergence=pv_divergence,
|
|
momentum_score=momentum_score,
|
|
)
|
|
|
|
def is_breakout(self, metrics: VolatilityMetrics) -> bool:
|
|
"""Determine if a stock is experiencing a breakout.
|
|
|
|
Args:
|
|
metrics: Volatility metrics for the stock
|
|
|
|
Returns:
|
|
True if breakout conditions are met
|
|
"""
|
|
return (
|
|
metrics.price_change_1m >= self.min_price_change
|
|
and metrics.volume_surge >= self.min_volume_surge
|
|
and metrics.pv_divergence > 0 # Bullish divergence
|
|
)
|
|
|
|
def is_breakdown(self, metrics: VolatilityMetrics) -> bool:
|
|
"""Determine if a stock is experiencing a breakdown.
|
|
|
|
Args:
|
|
metrics: Volatility metrics for the stock
|
|
|
|
Returns:
|
|
True if breakdown conditions are met
|
|
"""
|
|
return (
|
|
metrics.price_change_1m <= -self.min_price_change
|
|
and metrics.volume_surge >= self.min_volume_surge
|
|
and metrics.pv_divergence < 0 # Bearish divergence
|
|
)
|