yirugi
/
fithia2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: expand data pipeline and integrate event features into scoring

Browse Source 
Data expansion:
- Poll SEC 8-K filings from 2025-10-01 to 2026-03-12 (was ~2 months)
- Pipeline: 36 new filings → 24 new events → total 44 events, 15 symbols
- Re-export with merged features (market_v1+event_v1+financial_v1)
- Parquet columns: 7 → 41 (adds signal_strength, guidance_direction,
  document_quality, oneoff_penalty, eps_growth_qoq, etc.)

Score model v2:
- Add event quality component (15% weight): signal_strength_score,
  guidance_direction_score, document_quality_score
- Add risk penalty component (10% weight): inverted oneoff_penalty
- Rebalance market weights: reaction 25%, close 25%, volume 15%, gap 10%
- Graceful degradation when event features are absent (returns 0.5)

Export pipeline:
- Add --feature-versions CLI flag to merge multiple feature types
- export_dataset_snapshot() accepts feature_versions list parameter
- Groups features by event_id and merges feature_json dicts

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
main
I Luk Kim 5 months ago
parent 354f7a1716
commit d4ae900286
5 changed files with 261 additions and 50 deletions
Show Stats Download Patch File Download Diff File

9
apps/pipeline/dataset_export/main.py
Unescape Escape View File

@ -17,6 +17,7 @@ async def run_dataset_export(
snapshot_id: str | None,
split_policy: str,
output_dir: str,
feature_versions: list[str] | None = None,
) -> dict:
async with get_session() as session:
manifest = await export_dataset_snapshot(
@ -24,6 +25,7 @@ async def run_dataset_export(
snapshot_id=snapshot_id,
split_policy=split_policy,
output_dir=output_dir,
feature_versions=feature_versions,
)
return manifest
@ -41,6 +43,12 @@ def main() -> None:
default="./data/datasets/snapshots",
help="Output directory for Parquet files",
)
parser.add_argument(
"--feature-versions",
nargs="+",
default=None,
help="Feature versions to merge (e.g. market_v1 event_v1 financial_v1)",
)
parser.add_argument("--json", action="store_true", help="Print manifest JSON to stdout")
args = parser.parse_args()
@ -52,6 +60,7 @@ def main() -> None:
snapshot_id=args.snapshot_id,
split_policy=args.split_policy,
output_dir=args.output_dir,
feature_versions=args.feature_versions,
)
)

24
configs/experiments/expanded_scored_v1.json
Unescape Escape View File

@ -0,0 +1,24 @@
{
"experiment_name": "expanded_scored_v1",
"dataset_snapshot_id": "e684ab2c-cfd9-4d72-94de-f7cc497211b8",
"description": "Expanded dataset (44 events, Oct 2025 - Mar 2026) with merged market+event+financial features and rule-based scoring.",
"base_config": "configs/backtest/defaults.json",
"overrides": {
"signal": {
"score_threshold": 0.5,
"max_candidates_per_day": 10
},
"risk": {
"per_trade_risk_pct": 0.01,
"max_daily_new_risk_pct": 0.05,
"max_positions": 10,
"max_positions_per_sector": 5
},
"execution": {
"max_holding_days": 5
}
},
"splits": [],
"tags": ["expanded", "scored", "merged-features"],
"notes": "44 events from 15 symbols, 6 months of 8-K filings. Features: market_v1+event_v1+financial_v1."
}

157
libs/backtest/scoring.py
Unescape Escape View File

@ -1,18 +1,19 @@
"""Rule-based entry score model for the backtester.
Computes a composite score in [0, 1] from market features available
at event time (no forward-looking data). Higher score = more favorable
entry conditions for a long swing trade.
Computes a composite score in [0, 1] from market and event features
available at entry time (no forward-looking data). Higher score = more
favorable entry conditions for a long swing trade.
Based on Post-Earnings Announcement Drift (PEAD) microstructure:
- Moderate positive reaction → likely continuation
Market features (75% weight — primary signal):
- Moderate positive reaction → likely PEAD continuation
- Extreme positive reaction → already priced in, mean-reversion risk
- Close near session high → buyers in control
- Above-average volume → conviction (but extreme volume can signal exhaustion)
- Above-average volume → conviction (but extreme volume = exhaustion)
- Small positive gap → orderly strength
Design note: With limited data (14 records), this model uses general
market microstructure principles rather than fitted parameters. It can
be upgraded to ML when more training data is available.
Event features (25% weight — supplementary signal):
- Document quality & signal strength → confidence in the event parsing
- Risk flags (oneoff_penalty) → penalty for suspicious events
"""
from __future__ import annotations
@ -24,44 +25,60 @@ logger = get_logger(__name__)
def compute_entry_score(row: dict[str, Any]) -> float:
"""Compute composite entry score from market features.
"""Compute composite entry score from market + event features.
Components and weights:
1. Reaction quality (35%) — moderate positive return is ideal
2. Close strength (30%) — close near high = buyers won the day
3. Volume conviction (20%) — above-average but not exhaustion
4. Gap quality (15%) — small positive gap = orderly strength
Market (75%):
1. Reaction quality (25%) — moderate positive return is ideal
2. Close strength (25%) — close near high = buyers won the day
3. Volume conviction (15%) — above-average but not exhaustion
4. Gap quality (10%) — small positive gap = orderly strength
Event (25%):
5. Event quality (15%) — parser confidence + signal strength
6. Risk penalty (10%) — oneoff risk flags reduce score
Returns float in [0.0, 1.0].
"""
# Market components
reaction = _reaction_score(row)
close = _close_strength_score(row)
volume = _volume_score(row)
gap = _gap_score(row)
raw = reaction * 0.35 + close * 0.30 + volume * 0.20 + gap * 0.15
# Event components (gracefully handle missing features)
event = _event_quality_score(row)
risk = _risk_penalty_score(row)
raw = (
reaction * 0.25
+ close * 0.25
+ volume * 0.15
+ gap * 0.10
+ event * 0.15
+ risk * 0.10
)
return max(0.0, min(1.0, raw))
# ---------------------------------------------------------------------------
# Component scoring functions
# Market feature scoring
# ---------------------------------------------------------------------------
def _reaction_score(row: dict[str, Any]) -> float:
"""Score based on reaction_day_return.
Sweet spot: moderate positive return (0.5–3%) suggests post-event
Sweet spot: moderate positive return (0.5-3%) suggests post-event
continuation without being "already priced in".
Mapping:
+0.5% to +3% → 0.9 (ideal PEAD zone)
+0% to +0.5% → 0.65 (flat, uncertain direction)
+3% to +8% → 0.45 (getting priced in)
> +8% → 0.2 (extreme — mean reversion risk)
-2% to 0% → 0.4 (mild negative)
-5% to -2% → 0.25 (moderate negative)
< -5% → 0.1 (strongly bearish)
+0.5% to +3% -> 0.9 (ideal PEAD zone)
+0% to +0.5% -> 0.65 (flat, uncertain direction)
+3% to +8% -> 0.45 (getting priced in)
> +8% -> 0.2 (extreme — mean reversion risk)
-2% to 0% -> 0.4 (mild negative)
-5% to -2% -> 0.25 (moderate negative)
< -5% -> 0.1 (strongly bearish)
"""
rdr = row.get("reaction_day_return")
if rdr is None:
@ -87,7 +104,7 @@ def _reaction_score(row: dict[str, Any]) -> float:
def _close_strength_score(row: dict[str, Any]) -> float:
"""Score based on close_location [0=low, 1=high].
Linear mapping: 0.0 → 0.1, 1.0 → 1.0.
Linear mapping: 0.0 -> 0.1, 1.0 -> 1.0.
Close near session high = buyers controlled the day.
"""
cl = row.get("close_location")
@ -102,14 +119,14 @@ def _volume_score(row: dict[str, Any]) -> float:
"""Score based on volume_ratio_20d.
Above-average volume confirms conviction, but extreme volume
(>3×) can signal exhaustion or panic, so it gets discounted.
(>3x) can signal exhaustion or panic, so it gets discounted.
Mapping:
1.2×–2.0× → 0.8 (healthy conviction)
1.0×–1.2× → 0.6 (normal)
2.0×–3.0× → 0.55 (high — possible exhaustion)
> 3.0× → 0.4 (extreme — likely exhaustion)
< 1.0× → 0.3 (below average — no conviction)
1.2x-2.0x -> 0.8 (healthy conviction)
1.0x-1.2x -> 0.6 (normal)
2.0x-3.0x -> 0.55 (high — possible exhaustion)
> 3.0x -> 0.4 (extreme — likely exhaustion)
< 1.0x -> 0.3 (below average — no conviction)
"""
vr = row.get("volume_ratio_20d")
if vr is None:
@ -131,17 +148,17 @@ def _volume_score(row: dict[str, Any]) -> float:
def _gap_score(row: dict[str, Any]) -> float:
"""Score based on gap_size (open vs previous close).
Small positive gap (0–2%) = orderly bullish opening.
Small positive gap (0-2%) = orderly bullish opening.
Large gap (>5%) = potential exhaustion gap.
Negative gap = bearish opening pressure.
Mapping:
+0.5% to +2% → 0.8 (orderly strength)
0% to +0.5% → 0.6 (neutral-to-mild)
+2% to +5% → 0.5 (getting extended)
> +5% → 0.3 (exhaustion gap risk)
-2% to 0% → 0.4 (mild weakness)
< -2% → 0.2 (bearish gap)
+0.5% to +2% -> 0.8 (orderly strength)
0% to +0.5% -> 0.6 (neutral-to-mild)
+2% to +5% -> 0.5 (getting extended)
> +5% -> 0.3 (exhaustion gap risk)
-2% to 0% -> 0.4 (mild weakness)
< -2% -> 0.2 (bearish gap)
"""
gs = row.get("gap_size")
if gs is None:
@ -160,3 +177,67 @@ def _gap_score(row: dict[str, Any]) -> float:
return 0.4
else: # g < -0.02
return 0.2
# ---------------------------------------------------------------------------
# Event feature scoring
# ---------------------------------------------------------------------------
def _event_quality_score(row: dict[str, Any]) -> float:
"""Score based on event parsing quality and signal strength.
Combines:
- document_quality_score [0-1]: parser confidence in the extraction
- signal_strength_score [0-1]: strength of business fundamentals signals
- guidance_direction_score: 1.0=raised, 0.5=inline, 0.25=unclear, 0.0=lowered
When event features are absent (market_v1 only Parquet), returns 0.5.
"""
doc_q = row.get("document_quality_score")
sig_s = row.get("signal_strength_score")
guid = row.get("guidance_direction_score")
# If no event features at all, return neutral
if doc_q is None and sig_s is None and guid is None:
return 0.5
# Weight: signal_strength 40%, guidance 35%, document_quality 25%
scores = []
weights = []
if sig_s is not None:
scores.append(float(sig_s))
weights.append(0.40)
if guid is not None:
scores.append(float(guid))
weights.append(0.35)
if doc_q is not None:
scores.append(float(doc_q))
weights.append(0.25)
if not scores:
return 0.5
total_weight = sum(weights)
return sum(s * w for s, w in zip(scores, weights)) / total_weight
def _risk_penalty_score(row: dict[str, Any]) -> float:
"""Score based on oneoff_penalty (risk flags).
oneoff_penalty [0-1]: fraction of active risk flags.
Higher penalty = lower score (more risk = less favorable entry).
Inverted: 0.0 penalty -> 0.9 score, 1.0 penalty -> 0.2 score.
When absent, returns neutral 0.5.
"""
penalty = row.get("oneoff_penalty")
if penalty is None:
return 0.5
p = max(0.0, min(1.0, float(penalty)))
# Linear inversion: 0 -> 0.9, 1.0 -> 0.2
return 0.9 - 0.7 * p

32
libs/export/snapshot_export.py
Unescape Escape View File

@ -86,6 +86,7 @@ async def export_dataset_snapshot(
split_policy: str,
output_dir: str | Path,
feature_version: str = "market_v1",
feature_versions: list[str] | None = None,
label_version: str = "label-1.0.0",
parser_version: str = "rule-1.0.0",
) -> dict[str, Any]:
@ -96,7 +97,8 @@ async def export_dataset_snapshot(
snapshot_id: Unique ID for this snapshot (generated if None).
split_policy: Temporal split policy string (e.g. "temporal_70_15_15").
output_dir: Root directory for output files.
feature_version: Snapshot name filter for FeatureSnapshot.
feature_version: Snapshot name filter (used when feature_versions is None).
feature_versions: Merge multiple feature types (e.g. ["market_v1", "event_v1"]).
label_version: Label version filter for EventLabel.
parser_version: Parser version filter for Event.
@ -108,14 +110,17 @@ async def export_dataset_snapshot(
if snapshot_id is None:
snapshot_id = str(uuid.uuid4())
versions = feature_versions or [feature_version]
out_path = Path(output_dir) / snapshot_id
out_path.mkdir(parents=True, exist_ok=True)
# Query: JOIN feature_snapshots + event_labels via event_id
# When merging multiple versions, query all and group by event_id
stmt = (
select(FeatureSnapshot, EventLabel)
.join(EventLabel, FeatureSnapshot.event_id == EventLabel.event_id)
.where(FeatureSnapshot.snapshot_name == feature_version)
.where(FeatureSnapshot.snapshot_name.in_(versions))
.where(EventLabel.label_version == label_version)
.where(EventLabel.label_status == "ok")
.where(EventLabel.invalid_event_for_labeling.is_(False))
@ -124,13 +129,24 @@ async def export_dataset_snapshot(
result = await session.execute(stmt)
pairs = result.all()
rows: list[dict[str, Any]] = []
# Group features by event_id, merge feature_json from all versions
event_features: dict[str, dict[str, Any]] = {}
event_labels: dict[str, Any] = {}
for fs, lbl in pairs:
eid = fs.event_id
if eid not in event_features:
event_features[eid] = {}
event_labels[eid] = lbl
event_features[eid].update(fs.feature_json)
rows: list[dict[str, Any]] = []
for eid, features in event_features.items():
lbl = event_labels[eid]
row: dict[str, Any] = {
"event_id": fs.event_id,
"snapshot_name": fs.snapshot_name,
"snapshot_version": fs.snapshot_version,
**fs.feature_json,
"event_id": eid,
"snapshot_name": "+".join(versions),
"snapshot_version": "1.0.0",
**features,
"entry_convention": lbl.entry_convention,
"reaction_date": lbl.reaction_date.isoformat() if lbl.reaction_date else None,
"entry_date": lbl.entry_date.isoformat() if lbl.entry_date else None,
@ -169,7 +185,7 @@ async def export_dataset_snapshot(
"snapshot_id": snapshot_id,
"created_at_utc": utc_now().isoformat(),
"code_commit_hash": _get_git_commit_hash(),
"feature_version": feature_version,
"feature_version": "+".join(versions),
"parser_version": parser_version,
"label_version": label_version,
"split_policy": split_policy,

89
tests/unit/backtest/test_scoring.py
Unescape Escape View File

@ -5,8 +5,10 @@ import pytest
from libs.backtest.scoring import (
_close_strength_score,
_event_quality_score,
_gap_score,
_reaction_score,
_risk_penalty_score,
_volume_score,
compute_entry_score,
)
@ -136,7 +138,7 @@ class TestComputeEntryScore:
"gap_size": 0.01, # orderly → 0.8
}
score = compute_entry_score(row)
assert score > 0.8
assert score > 0.7
def test_bearish_setup_scores_low(self):
"""Negative return + close near low + below avg volume."""
@ -147,7 +149,7 @@ class TestComputeEntryScore:
"gap_size": -0.03, # bearish gap → 0.2
}
score = compute_entry_score(row)
assert score < 0.3
assert score < 0.35
def test_extreme_positive_penalized(self):
"""Very large positive reaction should be penalized."""
@ -210,6 +212,85 @@ class TestComputeEntryScore:
"volume_ratio_20d": 1.45,
"gap_size": 0.006,
})
assert aapl > 0.6, f"AAPL should be above 0.6, got {aapl:.3f}"
assert tsla < 0.4, f"TSLA should be below 0.4, got {tsla:.3f}"
assert aapl > 0.55, f"AAPL should be above 0.55, got {aapl:.3f}"
assert tsla < 0.45, f"TSLA should be below 0.45, got {tsla:.3f}"
assert aapl > tsla
def test_event_features_boost_score(self):
"""Strong event features should boost overall score."""
market_only = compute_entry_score({
"reaction_day_return": 0.01,
"close_location": 0.6,
"volume_ratio_20d": 1.5,
"gap_size": 0.01,
})
with_events = compute_entry_score({
"reaction_day_return": 0.01,
"close_location": 0.6,
"volume_ratio_20d": 1.5,
"gap_size": 0.01,
"signal_strength_score": 0.8,
"guidance_direction_score": 1.0,
"document_quality_score": 0.7,
"oneoff_penalty": 0.0,
})
assert with_events > market_only
def test_high_risk_penalty_lowers_score(self):
"""High oneoff_penalty should lower overall score."""
low_risk = compute_entry_score({
"reaction_day_return": 0.01,
"close_location": 0.6,
"oneoff_penalty": 0.0,
})
high_risk = compute_entry_score({
"reaction_day_return": 0.01,
"close_location": 0.6,
"oneoff_penalty": 1.0,
})
assert low_risk > high_risk
class TestEventQualityScore:
"""Event quality scoring."""
def test_strong_signals(self):
score = _event_quality_score({
"signal_strength_score": 0.8,
"guidance_direction_score": 1.0,
"document_quality_score": 0.7,
})
assert score > 0.7
def test_weak_signals(self):
score = _event_quality_score({
"signal_strength_score": 0.0,
"guidance_direction_score": 0.0,
"document_quality_score": 0.3,
})
assert score < 0.15
def test_missing_returns_neutral(self):
assert _event_quality_score({}) == 0.5
def test_partial_features(self):
"""Works with only some event features present."""
score = _event_quality_score({"signal_strength_score": 0.8})
assert score == pytest.approx(0.8, abs=0.01)
class TestRiskPenaltyScore:
"""Risk penalty scoring."""
def test_no_risk(self):
assert _risk_penalty_score({"oneoff_penalty": 0.0}) == pytest.approx(0.9)
def test_max_risk(self):
assert _risk_penalty_score({"oneoff_penalty": 1.0}) == pytest.approx(0.2)
def test_moderate_risk(self):
score = _risk_penalty_score({"oneoff_penalty": 0.5})
assert 0.4 < score < 0.7
def test_missing_returns_neutral(self):
assert _risk_penalty_score({}) == 0.5

Write Preview
Loading…
Cancel
Save