62b1a1f37a
Some checks failed
CI / test (pull_request) Has been cancelled
Implements issue #20 - Behavioral Rule: Volatility Hunter Components: 1. src/analysis/volatility.py - VolatilityAnalyzer with ATR calculation - Price change tracking (1m, 5m, 15m intervals) - Volume surge detection (ratio vs average) - Price-volume divergence analysis - Momentum scoring (0-100 scale) - Breakout/breakdown detection 2. src/analysis/scanner.py - MarketScanner for real-time stock scanning - Scans all available stocks every 60 seconds - Ranks by momentum score - Identifies top 5 movers per market - Dynamic watchlist updates 3. Integration with src/main.py - Auto-adjust WATCHLISTS dynamically - Replace laggards with leaders (max 2 per scan) - Volume confirmation required - Integrated with Context Tree L7 (real-time layer) 4. Comprehensive tests - 22 tests in tests/test_volatility.py - 99% coverage for analysis module - Tests for all volatility calculations - Tests for scanner ranking and watchlist updates - All tests passing Key Features: - Scan ALL stocks, not just current watchlist - Dynamic watchlist that adapts to market leaders - Context Tree integration for real-time data storage - Breakout detection with volume confirmation - Multi-timeframe momentum analysis Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
326 lines
10 KiB
Python
326 lines
10 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_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
|
|
)
|