Skip to content

PR #564: Wire arm angle + swing path + chase discipline into prop enrichment #433

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jaayslaughter-cpu merged 1 commit into from
May 15, 2026
+138 −11
Merged

PR #564: Wire arm angle + swing path + chase discipline into prop enrichment #433

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
109 changes: 99 additions & 10 deletions prop_enrichment_layer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1605,6 +1605,35 @@ def _dampen(base_prob_pct, adjustments, **kw): # noqa: E731
player, _k_sig, _k_sig_nudge,
)

# ── Arm angle deception signal (pitcher K props) ─────────────────────────
# Pitchers with extreme arm angles (very low/submarine or very high/overhand)
# generate more deception → more swing-and-miss vs. league-average arm slots.
# Signal: deviation from ~42° MLB average; max ±3pp adjustment.
if prop_type == "strikeouts":
_arm_pid = prop.get("player_id") or prop.get("mlbam_id")
if _arm_pid:
try:
from statcast_static_layer import get_pitcher_arm_angle as _gaa # noqa: PLC0415
_arm_data = _gaa(int(_arm_pid))
if _arm_data:
_ball_angle = float(_arm_data.get("ball_angle") or 0)
if _ball_angle > 0:
prop["_ball_angle"] = round(_ball_angle, 1)
# Deviation from 42° MLB average → deception premium
_dev = abs(_ball_angle - 42.0)
# <20° = submarine (sidearmer) = +2-3pp deception bonus
# 20-30° deviation = moderate deception = +1-2pp
# <10° deviation = generic arm slot = no signal
_arm_k_adj = round(min(0.030, max(0.0, (_dev - 8.0) / 22.0 * 0.030)), 4)
if _arm_k_adj >= 0.005:
prop["_arm_angle_adj"] = _arm_k_adj
logger.debug(
"[Enrichment] %s arm_angle=%.1f dev=%.1f arm_k_adj=+%.3f",
player, _ball_angle, _dev, _arm_k_adj,
)
except Exception as _aa_err:
logger.debug("[Enrichment] arm_angle skipped for %s: %s", player, _aa_err)

# ── TTOP: Times Through Order Penalty (pitcher K props) ────────────────
# Approximates expected TTO from l5_ip if not already provided by hub.
# _tto_k_adj is read by tasklets.py and added to raw_p before EV calc.
Expand Down Expand Up @@ -1639,6 +1668,60 @@ def _dampen(base_prob_pct, adjustments, **kw): # noqa: E731
except Exception as _bpv_err:
logger.debug("[Enrichment] BPV skipped for %s: %s", player, _bpv_err)

# ── Swing path K-susceptibility (bat-tracking-swing-path.csv) ──────
# swing_tilt: high = more uppercut = more Ks (>25° is above avg)
# ideal_attack_angle_rate: low = poor mechanics = more Ks (<60% is weak)
if _b_id:
try:
from statcast_static_layer import get_batter_bat_tracking as _gbt # noqa: PLC0415
_bt = _gbt(int(_b_id))
if _bt:
_tilt = float(_bt.get("swing_tilt") or 0)
_ideal = float(_bt.get("ideal_attack_angle_rate") or 0)
_swing_path_adj = 0.0
if _tilt > 0:
prop["_swing_tilt"] = round(_tilt, 2)
# League avg ~22°; +2.5pp per 12° above average
_swing_path_adj += (_tilt - 22.0) / 12.0 * 0.025
if _ideal > 0:
prop["_ideal_attack_rate"] = round(_ideal, 3)
# League avg ~0.62; low ideal rate = bad mechanics = more Ks
_swing_path_adj += (0.62 - _ideal) / 0.15 * 0.015
_swing_path_adj = round(max(-0.030, min(0.040, _swing_path_adj)), 4)
if abs(_swing_path_adj) >= 0.005:
prop["_swing_path_k_adj"] = _swing_path_adj
logger.debug(
"[Enrichment] %s swing_path_k_adj=%.3f (tilt=%.1f ideal=%.2f)",
player, _swing_path_adj, _tilt, _ideal,
)
except Exception as _sp_err:
logger.debug("[Enrichment] swing_path skipped for %s: %s", player, _sp_err)

# ── Chase zone discipline (swing-take.csv) ─────────────────────────
# runs_chase_pa < 0 = losing runs on chases = bad discipline = more Ks
# runs_chase_pa > 0 = disciplined = fewer Ks
if _b_id:
try:
from statcast_static_layer import get_batter_chase_discipline as _gcd # noqa: PLC0415
_chase_data = _gcd(int(_b_id))
if _chase_data:
_runs_chase_pa = _chase_data.get("runs_chase_pa")
if _runs_chase_pa is not None:
prop["_chase_runs_pa"] = round(_runs_chase_pa, 4)
# -0.04 runs/PA (very undisciplined) → +1.5pp Ks
# +0.04 runs/PA (disciplined) → -1.5pp Ks
_chase_k_adj = round(
max(-0.025, min(0.025, -_runs_chase_pa / 0.04 * 0.015)), 4
)
if abs(_chase_k_adj) >= 0.005:
prop["_chase_discipline_k_adj"] = _chase_k_adj
logger.debug(
"[Enrichment] %s chase_discipline_k_adj=%.3f (runs/PA=%.4f)",
player, _chase_k_adj, _runs_chase_pa,
)
except Exception as _cd_err:
logger.debug("[Enrichment] chase_discipline skipped for %s: %s", player, _cd_err)

# ── Defense OAA (batter props) ─────────────────────────────────────────
if is_batter_prop:
try:
Expand Down Expand Up @@ -1870,15 +1953,18 @@ def _dampen(base_prob_pct, adjustments, **kw): # noqa: E731
# +4pp chase + +6pp arsenal = +18pp raw → ~+13pp dampened).
_all_adjs: list = []
for _adj_key, _adj_label in [
("_bayesian_nudge", "bayesian"),
("_cv_nudge", "cv_consistency"),
("_form_adj", "mlb_form"),
("_chase_k_adj", "lineup_chase"),
("_drama_penalty_pp", "bernoulli_drama"),
("_arsenal_k_adj", "arsenal_k_sig"),
("_ump_k_adj", "umpire"),
("_steamer_adj", "steamer"),
("_tto_k_adj", "ttop_decay"),
("_bayesian_nudge", "bayesian"),
("_cv_nudge", "cv_consistency"),
("_form_adj", "mlb_form"),
("_chase_k_adj", "lineup_chase"),
("_drama_penalty_pp", "bernoulli_drama"),
("_arsenal_k_adj", "arsenal_k_sig"),
("_ump_k_adj", "umpire"),
("_steamer_adj", "steamer"),
("_tto_k_adj", "ttop_decay"),
("_arm_angle_adj", "arm_angle_deception"),
("_swing_path_k_adj", "swing_path_k"),
("_chase_discipline_k_adj", "chase_discipline_k"),
]:
_v = float(prop.get(_adj_key, 0.0) or 0.0)
if _v != 0.0:
Expand Down Expand Up @@ -1943,7 +2029,10 @@ def _dampen(base_prob_pct, adjustments, **kw): # noqa: E731
"market_flag": str(prop.get("_market_flag", "CLEAN")),
"injury": round(float(prop.get("_injury_confidence_penalty", 0) or 0), 3),
"park": round(float(prop.get("_park_k_factor", 1.0) or 1.0), 3),
"lambda_bias": round(float(prop.get("_lambda_bias", 0.0) or 0.0), 4),
"lambda_bias": round(float(prop.get("_lambda_bias", 0.0) or 0.0), 4),
"arm_angle": round(float(prop.get("_arm_angle_adj", 0.0) or 0.0), 4),
"swing_path_k": round(float(prop.get("_swing_path_k_adj", 0.0) or 0.0), 4),
"chase_disc_k": round(float(prop.get("_chase_discipline_k_adj",0.0) or 0.0), 4),
}

enriched_count += 1
Expand Down
40 changes: 39 additions & 1 deletion statcast_static_layer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,8 @@

# Spin direction (2026)
_spin_direction: dict[tuple, dict] = {} # (player_id_int, api_pitch_type) → spin stats
_pitcher_arm_angles: dict[int, dict] = {} # pitcher_arm_angles.csv — ball_angle, pitch_hand, release_z
_pitcher_arm_angles: dict[int, dict] = {} # pitcher_arm_angles.csv
_swing_take: dict[int, dict] = {} # swing-take.csv — chase/shadow/waste zone runs


# ── Helpers ───────────────────────────────────────────────────────────────────
Expand Down Expand Up @@ -537,6 +538,26 @@
}
logger.info("[StatcastStatic] pitcher_arm_angles: %d pitchers loaded", len(_pitcher_arm_angles))

# ── Batter swing-take discipline (swing-take.csv) ────────────────────────
# runs_heart/shadow/chase/waste: run value gained/lost per zone decision
# Negative runs_chase = losing runs from chasing = more Ks expected
for r in _read_csv("swing-take.csv"):
try:
pid = int(r.get("player_id", 0) or 0)
if not pid:
continue
pa = max(1, int(r.get("pa", 1) or 1))
_swing_take[pid] = {
"runs_chase_pa": round(_safe_float(r.get("runs_chase")) / pa, 5) if r.get("runs_chase") else None,
"runs_shadow_pa": round(_safe_float(r.get("runs_shadow")) / pa, 5) if r.get("runs_shadow") else None,
"runs_heart_pa": round(_safe_float(r.get("runs_heart")) / pa, 5) if r.get("runs_heart") else None,
"runs_waste_pa": round(_safe_float(r.get("runs_waste")) / pa, 5) if r.get("runs_waste") else None,
Comment on lines +551 to +554
Copy link
Copy Markdown

@gemini-code-assist gemini-code-assist Bot May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

The current implementation of _safe_float (line 111) returns the default value (which is None here) when the input is 0.0. This causes a TypeError during the division by pa on lines 551-554, which is then caught by the except block on line 557, resulting in the entire player record being skipped. Since 0.0 is a valid and common run value in the Statcast dataset, it should be handled correctly. Using float() directly is safe here because the if condition already ensures the value is a non-empty string.

Suggested change
"runs_chase_pa": round(_safe_float(r.get("runs_chase")) / pa, 5) if r.get("runs_chase") else None,
"runs_shadow_pa": round(_safe_float(r.get("runs_shadow")) / pa, 5) if r.get("runs_shadow") else None,
"runs_heart_pa": round(_safe_float(r.get("runs_heart")) / pa, 5) if r.get("runs_heart") else None,
"runs_waste_pa": round(_safe_float(r.get("runs_waste")) / pa, 5) if r.get("runs_waste") else None,
"runs_chase_pa": round(float(r.get("runs_chase")) / pa, 5) if r.get("runs_chase") else None,
"runs_shadow_pa": round(float(r.get("runs_shadow")) / pa, 5) if r.get("runs_shadow") else None,
"runs_heart_pa": round(float(r.get("runs_heart")) / pa, 5) if r.get("runs_heart") else None,
"runs_waste_pa": round(float(r.get("runs_waste")) / pa, 5) if r.get("runs_waste") else None,

"pa": pa,
}
except (ValueError, TypeError, ZeroDivisionError):
continue
logger.info("[StatcastStatic] swing_take: %d batters loaded", len(_swing_take))

# ── Bat tracking swing path (bat-tracking-swing-path.csv) ────────────
# Supplements _batter_tracking with attack_angle, swing_tilt, ideal_attack_rate
_swing_path_count = 0
Expand Down Expand Up @@ -940,6 +961,23 @@
return _pitcher_arm_angles.get(int(player_id), {})


def get_batter_chase_discipline(player_id: int) -> dict:
"""Return batter swing-take zone discipline data.

Keys: runs_chase_pa, runs_shadow_pa, runs_heart_pa, runs_waste_pa, pa
runs_chase_pa < 0 → losing runs on chases → high K susceptibility
runs_chase_pa > 0 → disciplined or good contact on chase pitches

Falls back to empty dict if player not in dataset.
"""
if not _LOADED:

Check warning on line 973 in statcast_static_layer.py

View check run for this annotation

Codacy Production / Codacy Static Code Analysis

statcast_static_layer.py#L973 

undefined name '_LOADED' (F821)
Copy link
Copy Markdown

@deepsource-io deepsource-io Bot May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Undefined variable '_LOADED'


The variable name is not defined where it is used.
This will lead to an error during the runtime.
Make sure there is no typo. If the name was supposed to be imported, verify that you've actually imported the name.

_load()
Comment on lines +973 to +974
Copy link
Copy Markdown

@gemini-code-assist gemini-code-assist Bot May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

The variable _LOADED is not defined in this module; the correct variable name is _loaded (lowercase), as seen on line 68. Furthermore, the _load() function already contains an internal check for the _loaded state (line 137), making this manual check redundant. Calling _load() directly is the standard pattern used throughout this module.

    _load()

try:
return _swing_take.get(int(player_id), {})
except (ValueError, TypeError):
return {}


# -- Matchup --

def get_matchup_k_boost(pitcher_id: int, batter_id: int) -> float:
Expand Down