You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
"""Intraday volume profile features from 5-minute bars.
|
|
|
|
Computes institutional conviction signals from reaction-day intraday data:
|
|
- first_half_volume_pct: fraction of volume in 9:30-12:00
|
|
- volume_front_loading_ratio: first_half / second_half volume
|
|
- vwap_premium_pct: (close - vwap) / vwap
|
|
- institutional_conviction_score: composite [0, 1]
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from libs.common.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
# NYSE session: 9:30 ET - 16:00 ET; midpoint at 12:00 ET
|
|
_MIDPOINT_HOUR = 12
|
|
_MIDPOINT_MINUTE = 0
|
|
|
|
|
|
def compute_intraday_features(bars: list[dict[str, Any]]) -> dict[str, Any] | None:
|
|
"""Compute intraday volume profile features from 5-minute bars.
|
|
|
|
Args:
|
|
bars: list of dicts with keys: timestamp, open, high, low, close, volume.
|
|
timestamp format: "2024-01-15T09:30:00-05:00" or similar.
|
|
|
|
Returns:
|
|
Feature dict or None if insufficient data.
|
|
"""
|
|
if not bars or len(bars) < 10:
|
|
return None
|
|
|
|
first_half_vol = 0
|
|
second_half_vol = 0
|
|
total_volume = 0
|
|
vwap_numerator = 0.0
|
|
|
|
for bar in bars:
|
|
vol = int(bar.get("volume", 0))
|
|
if vol <= 0:
|
|
continue
|
|
|
|
ts = bar.get("timestamp", "")
|
|
hour, minute = _extract_hour_minute(ts)
|
|
if hour is None:
|
|
continue
|
|
|
|
total_volume += vol
|
|
typical_price = (float(bar["high"]) + float(bar["low"]) + float(bar["close"])) / 3.0
|
|
vwap_numerator += typical_price * vol
|
|
|
|
if hour < _MIDPOINT_HOUR or (hour == _MIDPOINT_HOUR and minute == 0):
|
|
first_half_vol += vol
|
|
else:
|
|
second_half_vol += vol
|
|
|
|
if total_volume < 100:
|
|
return None
|
|
|
|
first_half_pct = first_half_vol / total_volume if total_volume > 0 else 0.0
|
|
front_loading = first_half_vol / max(1, second_half_vol)
|
|
vwap = vwap_numerator / total_volume if total_volume > 0 else 0.0
|
|
|
|
last_close = float(bars[-1].get("close", 0.0))
|
|
vwap_premium = (last_close - vwap) / vwap if vwap > 0 else 0.0
|
|
|
|
conviction = _compute_conviction_score(first_half_pct, front_loading, vwap_premium)
|
|
|
|
return {
|
|
"first_half_volume_pct": round(first_half_pct, 4),
|
|
"volume_front_loading_ratio": round(front_loading, 4),
|
|
"vwap_premium_pct": round(vwap_premium, 6),
|
|
"institutional_conviction_score": round(conviction, 4),
|
|
}
|
|
|
|
|
|
def _compute_conviction_score(
|
|
first_half_pct: float,
|
|
front_loading: float,
|
|
vwap_premium: float,
|
|
) -> float:
|
|
"""Composite conviction score [0, 1].
|
|
|
|
High conviction = volume front-loaded (institutions acting early)
|
|
+ closing above VWAP (sustained buying pressure).
|
|
|
|
Components (equal weight):
|
|
- Front-loading: 1.4-2.0x first/second half ratio -> 0.5-1.0
|
|
- VWAP premium: 0%-2%+ close above VWAP -> 0.5-1.0
|
|
- Volume concentration: 55-70% in first half -> 0.5-1.0
|
|
"""
|
|
# Front-loading ratio score
|
|
if front_loading < 1.0:
|
|
fl_score = 0.2
|
|
elif front_loading < 1.4:
|
|
fl_score = 0.2 + (front_loading - 1.0) / 0.4 * 0.3
|
|
elif front_loading <= 2.0:
|
|
fl_score = 0.5 + (front_loading - 1.4) / 0.6 * 0.5
|
|
else:
|
|
fl_score = 1.0
|
|
|
|
# VWAP premium score
|
|
if vwap_premium < 0:
|
|
vwap_score = max(0.0, 0.3 + vwap_premium * 10)
|
|
elif vwap_premium < 0.01:
|
|
vwap_score = 0.3 + vwap_premium / 0.01 * 0.4
|
|
elif vwap_premium <= 0.02:
|
|
vwap_score = 0.7 + (vwap_premium - 0.01) / 0.01 * 0.3
|
|
else:
|
|
vwap_score = 1.0
|
|
|
|
# Volume concentration score
|
|
if first_half_pct < 0.45:
|
|
conc_score = 0.1
|
|
elif first_half_pct < 0.55:
|
|
conc_score = 0.1 + (first_half_pct - 0.45) / 0.10 * 0.4
|
|
elif first_half_pct <= 0.70:
|
|
conc_score = 0.5 + (first_half_pct - 0.55) / 0.15 * 0.5
|
|
else:
|
|
conc_score = 1.0
|
|
|
|
return max(0.0, min(1.0, (fl_score + vwap_score + conc_score) / 3.0))
|
|
|
|
|
|
def _extract_hour_minute(timestamp_str: str) -> tuple[int | None, int | None]:
|
|
"""Extract hour and minute from an ISO timestamp string."""
|
|
try:
|
|
# Handle formats like "2024-01-15T09:30:00-05:00" or "2024-01-15 09:30:00"
|
|
time_part = timestamp_str.split("T")[-1] if "T" in timestamp_str else timestamp_str.split(" ")[-1]
|
|
parts = time_part.split(":")
|
|
return int(parts[0]), int(parts[1])
|
|
except (IndexError, ValueError):
|
|
return None, None
|