scripts/screen_pead.py
#!/usr/bin/env python3
"""
PEAD Stock Screener - Main Orchestrator
Screens post-earnings gap-up stocks for Post-Earnings Announcement Drift (PEAD)
patterns using weekly candle analysis.
Two input modes:
Mode A: FMP earnings calendar -> profile batch -> gap filter -> weekly analysis
Mode B: earnings-trade-analyzer JSON output -> grade filter -> weekly analysis
Usage:
# Mode A: FMP earnings calendar (default)
python3 screen_pead.py --api-key YOUR_KEY --output-dir reports/
# Mode B: From earnings-trade-analyzer JSON
python3 screen_pead.py --candidates-json reports/earnings_analysis.json --output-dir reports/
Output:
- JSON: pead_screener_YYYY-MM-DD_HHMMSS.json
- Markdown: pead_screener_YYYY-MM-DD_HHMMSS.md
"""
import argparse
import json
import logging
import math
import os
import sys
from datetime import datetime, timedelta
from typing import Optional
# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(__file__))
from calculators.breakout_calculator import calculate_breakout
from calculators.liquidity_calculator import calculate_liquidity
from calculators.risk_reward_calculator import calculate_risk_reward
from calculators.weekly_candle_calculator import analyze_weekly_pattern, daily_to_weekly
from fmp_client import ApiCallBudgetExceeded, FMPClient
from report_generator import generate_json_report, generate_markdown_report
from scorer import calculate_composite_score
logger = logging.getLogger(__name__)
def calculate_price_gap(daily_prices: list[dict], earnings_date: str, timing: str) -> float:
"""Calculate actual price gap from daily OHLCV data.
BMO: gap = (open[earnings_date] / close[prev_day]) - 1
AMC/unknown: gap = (open[next_day] / close[earnings_date]) - 1
Args:
daily_prices: Most-recent-first daily price data
earnings_date: YYYY-MM-DD string
timing: 'bmo', 'amc', or 'unknown'
Returns:
Gap percentage (e.g. 6.3 for 6.3%), or 0.0 if calculation not possible.
"""
# Find earnings date index
earnings_idx = -1
for i, bar in enumerate(daily_prices):
if bar.get("date") == earnings_date:
earnings_idx = i
break
if earnings_idx == -1:
return 0.0
timing_lower = (timing or "").lower().strip()
if timing_lower == "bmo":
# BMO: gap = open[earnings_date] / close[prev_day] - 1
prev_idx = earnings_idx + 1 # most-recent-first
if prev_idx >= len(daily_prices):
return 0.0
base_price = daily_prices[prev_idx].get("close", 0)
gap_price = daily_prices[earnings_idx].get("open", 0)
else:
# AMC or unknown: gap = open[next_day] / close[earnings_date] - 1
next_idx = earnings_idx - 1 # most-recent-first
if next_idx < 0:
return 0.0
base_price = daily_prices[earnings_idx].get("close", 0)
gap_price = daily_prices[next_idx].get("open", 0)
if not base_price:
return 0.0
return round(((gap_price / base_price) - 1.0) * 100.0, 2)
def parse_arguments():
parser = argparse.ArgumentParser(
description="PEAD Stock Screener - Post-Earnings Announcement Drift"
)
# Common arguments
parser.add_argument(
"--api-key", help="FMP API key (defaults to FMP_API_KEY environment variable)"
)
parser.add_argument(
"--watch-weeks",
type=int,
default=5,
help="Monitoring period in weeks after earnings (default: 5)",
)
parser.add_argument(
"--max-api-calls",
type=int,
default=200,
help="API call budget (default: 200)",
)
parser.add_argument(
"--top",
type=int,
default=20,
help="Top results to include in report (default: 20)",
)
parser.add_argument(
"--output-dir",
default="reports/",
help="Output directory for reports (default: reports/)",
)
# Mode A arguments
parser.add_argument(
"--lookback-days",
type=int,
default=14,
help="Days back for earnings calendar (Mode A, default: 14)",
)
parser.add_argument(
"--min-gap",
type=float,
default=3.0,
help="Minimum earnings gap %% (Mode A, default: 3.0)",
)
parser.add_argument(
"--min-market-cap",
type=float,
default=500_000_000,
help="Minimum market cap (Mode A, default: 500000000)",
)
# Mode B arguments
parser.add_argument(
"--candidates-json",
help="Path to earnings-trade-analyzer JSON output (Mode B)",
)
parser.add_argument(
"--min-grade",
default="B",
choices=["A", "B", "C", "D"],
help="Minimum grade filter (Mode B, default: B)",
)
return parser.parse_args()
def validate_input_json(data: dict) -> list[dict]:
"""
Validate earnings-trade-analyzer JSON output for Mode B.
Checks:
1. schema_version == "1.0" -> ValueError if mismatch
2. 'results' key exists and is a list
3. Each result has required fields: symbol, earnings_date, earnings_timing, gap_pct, grade
4. Missing fields -> warn + skip that record (don't abort unless ALL fail)
Returns:
Validated list of result dicts.
Raises:
ValueError: If schema_version != "1.0" or all records are invalid
"""
# Check schema version
schema_version = data.get("schema_version", "")
if schema_version != "1.0":
raise ValueError(
f"Schema version mismatch: expected '1.0', got '{schema_version}'. "
"This input may be from an incompatible version of earnings-trade-analyzer."
)
# Check results key
results = data.get("results")
if not isinstance(results, list):
raise ValueError("Input JSON missing 'results' key or 'results' is not a list")
required_fields = ["symbol", "earnings_date", "earnings_timing", "gap_pct", "grade"]
valid_timings = {"bmo", "amc", "unknown"}
valid_grades = {"A", "B", "C", "D"}
validated = []
for i, record in enumerate(results):
# Check required field existence
missing = [f for f in required_fields if f not in record]
if missing:
logger.warning(
"Skipping record %d: missing required fields %s (has keys: %s)",
i,
missing,
list(record.keys()),
)
continue
# Type and value range validation
errors = []
if not isinstance(record["symbol"], str) or not record["symbol"].strip():
errors.append("symbol must be a non-empty string")
if not isinstance(record["earnings_date"], str) or len(record["earnings_date"]) != 10:
errors.append("earnings_date must be YYYY-MM-DD string")
if record["earnings_timing"] not in valid_timings:
errors.append(f"earnings_timing '{record['earnings_timing']}' not in {valid_timings}")
if not isinstance(record["gap_pct"], (int, float)):
errors.append(f"gap_pct must be numeric, got {type(record['gap_pct']).__name__}")
if record["grade"] not in valid_grades:
errors.append(f"grade '{record['grade']}' not in {valid_grades}")
if errors:
logger.warning(
"Skipping record %d (%s): %s",
i,
record.get("symbol", "?"),
"; ".join(errors),
)
continue
validated.append(record)
if not validated:
raise ValueError(
f"All {len(results)} records failed validation. No valid candidates to process."
)
return validated
def calculate_setup_quality(gap_pct: float, pattern_result: dict) -> float:
"""Calculate setup quality score based on earnings gap and pattern.
Args:
gap_pct: Earnings gap percentage
pattern_result: Result from analyze_weekly_pattern()
Returns:
Setup quality score (0-100)
"""
score = 0.0
# Gap quality (0-50 points)
if gap_pct >= 10.0:
score += 50
elif gap_pct >= 7.0:
score += 40
elif gap_pct >= 5.0:
score += 30
elif gap_pct >= 3.0:
score += 20
else:
score += 10
# Pattern quality (0-50 points)
stage = pattern_result.get("stage", "MONITORING")
weeks = pattern_result.get("weeks_since_earnings", 0)
red_candle = pattern_result.get("red_candle")
if stage == "BREAKOUT":
score += 50
elif stage == "SIGNAL_READY":
score += 40
# Bonus for red candle with long lower wick (institutional support)
if red_candle and red_candle.get("lower_wick_pct", 0) > 30:
score += 5
elif stage == "MONITORING":
# Earlier in the cycle is better
if weeks <= 2:
score += 25
else:
score += 15
else: # EXPIRED
score += 0
return min(100.0, score)
def analyze_stock(
symbol: str,
daily_prices: list[dict],
earnings_date: str,
earnings_timing: str,
gap_pct: float,
current_price: float,
watch_weeks: int = 5,
) -> Optional[dict]:
"""
Full PEAD analysis for a single stock.
Args:
symbol: Stock symbol
daily_prices: Most-recent-first daily OHLCV data
earnings_date: Earnings announcement date (YYYY-MM-DD)
earnings_timing: 'bmo', 'amc', or 'unknown'
gap_pct: Earnings gap percentage
current_price: Current stock price
watch_weeks: Maximum monitoring window in weeks
Returns:
Analysis result dict or None on failure
"""
if not daily_prices or len(daily_prices) < 5:
return None
# 1. Convert to weekly candles
weekly_candles = daily_to_weekly(daily_prices, earnings_date=earnings_date)
if not weekly_candles:
return None
# 2. Analyze weekly pattern
pattern = analyze_weekly_pattern(weekly_candles, earnings_date, watch_weeks=watch_weeks)
# 3. Calculate setup quality
setup_score = calculate_setup_quality(gap_pct, pattern)
# 4. Calculate breakout
red_candle = pattern.get("red_candle")
if red_candle:
breakout = calculate_breakout(weekly_candles, red_candle, current_price)
else:
breakout = {
"is_breakout": False,
"breakout_pct": 0.0,
"volume_confirmation": False,
"score": 0.0,
}
# 5. Calculate liquidity
liquidity = calculate_liquidity(daily_prices, current_price)
# 6. Calculate risk/reward
if red_candle:
rr = calculate_risk_reward(current_price, red_candle)
else:
rr = {
"entry_price": current_price,
"stop_price": 0.0,
"target_price": 0.0,
"risk_pct": 0.0,
"reward_pct": 0.0,
"risk_reward_ratio": 0.0,
"score": 25.0,
}
# 7. Composite score
composite = calculate_composite_score(
setup_score=setup_score,
breakout_score=breakout["score"],
liquidity_score=liquidity["score"],
rr_score=rr["score"],
)
return {
"symbol": symbol,
"stage": pattern["stage"],
"earnings_date": earnings_date,
"earnings_timing": earnings_timing,
"gap_pct": gap_pct,
"weeks_since_earnings": pattern["weeks_since_earnings"],
"red_candle": red_candle,
"current_price": current_price,
"breakout_pct": breakout["breakout_pct"],
"entry_price": rr["entry_price"],
"stop_price": rr["stop_price"],
"target_price": rr["target_price"],
"risk_pct": rr["risk_pct"],
"risk_reward_ratio": rr["risk_reward_ratio"],
"adv20_dollar": liquidity["adv20_dollar"],
"composite_score": composite["composite_score"],
"rating": composite["rating"],
"guidance": composite["guidance"],
"components": composite["component_breakdown"],
}
def main():
args = parse_arguments()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(levelname)s: %(message)s",
)
print("=" * 70)
print("PEAD Stock Screener")
print("Post-Earnings Announcement Drift")
print("=" * 70)
print()
# Determine mode
mode = "B" if args.candidates_json else "A"
print(f"Mode: {mode} ({'JSON Input' if mode == 'B' else 'FMP Earnings Calendar'})")
# Initialize FMP client (needed for both modes for historical data)
try:
client = FMPClient(api_key=args.api_key, max_api_calls=args.max_api_calls)
print("FMP API client initialized")
except ValueError as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
# ========================================================================
# Phase 1: Get Candidates
# ========================================================================
print()
print("Phase 1: Get Candidates")
print("-" * 70)
candidates = []
if mode == "A":
candidates, reason = _get_candidates_mode_a(client, args)
else:
candidates, reason = _get_candidates_mode_b(args)
if not candidates:
exit_code, level, message = _ZERO_RESULT_REASONS.get(
reason, (1, "ERROR", f"No candidates found (reason: {reason}).")
)
print(f"ZERO_RESULT_REASON={reason}", file=sys.stderr)
print(f" {level}: {message}", file=sys.stderr)
sys.exit(exit_code)
print(f" Total candidates: {len(candidates)}")
print()
# ========================================================================
# Phase 1.5: Budget Check
# ========================================================================
print("Phase 1.5: Budget Check")
print("-" * 70)
api_stats = client.get_api_stats()
remaining = api_stats["budget_remaining"]
needed = len(candidates) # 1 historical call per candidate
print(f" API calls remaining: {remaining}")
print(f" Estimated calls needed: {needed} (1 per candidate)")
if needed > remaining:
# Trim candidates to fit budget
candidates = candidates[:remaining]
print(f" WARNING: Trimmed to {len(candidates)} candidates to fit API budget")
else:
print(" Budget sufficient")
print()
# Timing diagnostics (Issue #352), counted AFTER the budget trim above so
# the population matches the candidates that actually enter Phase 2 (same
# convention as earnings-trade-analyzer). Mode A candidates carry a
# normalized earnings_timing from the FMP calendar; Mode B candidates
# carry whatever the input JSON already recorded, so leave the aggregate
# None rather than re-deriving a count that duplicates the upstream
# report's own metadata.
if mode == "A":
timing_candidates_total = len(candidates)
timing_unknown_count = sum(1 for c in candidates if c.get("earnings_timing") == "unknown")
timing_source = "fmp_stable_includeReportTimes"
else:
timing_candidates_total = None
timing_unknown_count = None
timing_source = None
# ========================================================================
# Phase 2: Fetch Historical Data & Weekly Candle Analysis
# ========================================================================
print("Phase 2: Fetch Historical Data")
print("-" * 70)
results = []
for i, candidate in enumerate(candidates):
symbol = candidate["symbol"]
if (i + 1) % 10 == 0 or i == len(candidates) - 1:
print(f" Progress: {i + 1}/{len(candidates)}", flush=True)
try:
data = client.get_historical_prices(symbol, days=90)
except ApiCallBudgetExceeded:
print(f" WARNING: API budget exceeded at {symbol}. Processing collected data.")
break
if not data or "historical" not in data:
continue
daily_prices = data["historical"]
if not daily_prices:
continue
current_price = daily_prices[0].get("close", 0)
if current_price <= 0:
continue
# Calculate actual price gap if not already provided (Mode A)
gap_pct = candidate.get("gap_pct")
if gap_pct is None:
gap_pct = calculate_price_gap(
daily_prices,
candidate["earnings_date"],
candidate.get("earnings_timing", ""),
)
# Apply min-gap filter (using actual price gap, not EPS estimate)
if mode == "A" and abs(gap_pct) < args.min_gap:
continue
# Run analysis
analysis = analyze_stock(
symbol=symbol,
daily_prices=daily_prices,
earnings_date=candidate["earnings_date"],
earnings_timing=candidate.get("earnings_timing", ""),
gap_pct=gap_pct,
current_price=current_price,
watch_weeks=args.watch_weeks,
)
if analysis:
print(
f" {symbol:6} Stage: {analysis['stage']:14} "
f"Score: {analysis['composite_score']:5.1f} ({analysis['rating']})"
)
results.append(analysis)
print()
# ========================================================================
# Phase 3: Score & Report
# ========================================================================
print("Phase 3: Generate Reports")
print("-" * 70)
# Create output directory if needed
os.makedirs(args.output_dir, exist_ok=True)
timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
json_file = os.path.join(args.output_dir, f"pead_screener_{timestamp}.json")
md_file = os.path.join(args.output_dir, f"pead_screener_{timestamp}.md")
api_stats = client.get_api_stats()
metadata = {
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"lookback_days": args.lookback_days if mode == "A" else None,
"watch_weeks": args.watch_weeks,
"mode": mode,
"input_file": args.candidates_json if mode == "B" else None,
"min_gap": args.min_gap if mode == "A" else None,
"min_market_cap": args.min_market_cap if mode == "A" else None,
"min_grade": args.min_grade if mode == "B" else None,
"api_stats": api_stats,
"timing_unknown_count": timing_unknown_count,
"timing_candidates_total": timing_candidates_total,
"timing_source": timing_source,
}
# Sort by stage priority then composite score before top-N cutoff
stage_priority = {"BREAKOUT": 0, "SIGNAL_READY": 1, "MONITORING": 2, "EXPIRED": 3}
results.sort(
key=lambda r: (stage_priority.get(r["stage"], 9), -r["composite_score"]),
)
top_results = results[: args.top] if len(results) > args.top else results
generate_json_report(top_results, metadata, json_file)
generate_markdown_report(top_results, metadata, md_file)
# ========================================================================
# Summary
# ========================================================================
print()
print("=" * 70)
print("PEAD Screening Complete")
print("=" * 70)
# Stage counts
stage_counts = {}
for r in results:
stage = r.get("stage", "UNKNOWN")
stage_counts[stage] = stage_counts.get(stage, 0) + 1
print()
print("Stage Distribution:")
for stage in ["BREAKOUT", "SIGNAL_READY", "MONITORING", "EXPIRED"]:
count = stage_counts.get(stage, 0)
print(f" {stage:14} {count}")
# Top 5
if results:
print()
print(f"Top {min(5, len(results))} Results:")
# Sort by stage priority then score
stage_priority = {"BREAKOUT": 0, "SIGNAL_READY": 1, "MONITORING": 2, "EXPIRED": 3}
sorted_results = sorted(
results,
key=lambda r: (stage_priority.get(r["stage"], 9), -r["composite_score"]),
)
for i, r in enumerate(sorted_results[:5], 1):
print(
f" {i}. {r['symbol']:6} {r['stage']:14} "
f"Score: {r['composite_score']:5.1f} ({r['rating']})"
)
else:
print()
print(" No PEAD candidates found.")
print()
print(f" JSON Report: {json_file}")
print(f" Markdown Report: {md_file}")
print()
print("API Usage:")
print(f" API calls made: {api_stats['api_calls_made']}")
print(f" Budget remaining: {api_stats['budget_remaining']}")
print()
def _coerce_market_cap(value) -> Optional[float]:
"""Return ``value`` as a number, or None when it is missing or non-numeric.
Native ints/floats pass through unchanged (so the JSON report keeps the
integer shape FMP sends); numeric strings are parsed; anything else is None.
"""
if value is None or isinstance(value, bool):
return None
if isinstance(value, (int, float)):
parsed = value
else:
try:
parsed = float(value)
except (TypeError, ValueError):
return None
# float("nan") / float("inf") parse without error but would bypass the
# `< min_market_cap` floor (both comparisons are False); reject them.
return parsed if math.isfinite(parsed) else None
def profile_market_cap(profile: dict) -> float:
"""Read market cap from an FMP profile dict.
``/stable/profile`` returns ``marketCap``; the legacy ``/api/v3/profile``
returned ``mktCap``. The stable key wins when it holds a usable number;
otherwise fall back to the legacy key. Missing / null / non-numeric values
collapse to 0.0 so the caller's ``<`` comparison never raises and the
symbol simply fails the cap floor (Issue #328).
"""
for key in ("marketCap", "mktCap"):
cap = _coerce_market_cap(profile.get(key))
if cap is not None:
return cap
return 0.0
# ZERO_RESULT_REASON -> (exit_code, level, one-line explanation). Both
# `_get_candidates_mode_a` and `_get_candidates_mode_b` return `(candidates,
# reason)`; `reason` is None when `candidates` is non-empty. See
# docs/dev/provider-contracts.md.
_ZERO_RESULT_REASONS = {
"no_earnings_rows": (
0,
"WARNING",
"The FMP earnings calendar returned no rows for the selected date range.",
),
"calendar_fetch_failed": (
1,
"ERROR",
"The earnings calendar fetch failed (no usable response body); "
"the provider may be down or the response shape may have changed.",
),
"profiles_budget_exhausted": (
0,
"WARNING",
"API budget was exhausted before any company profile could be fetched.",
),
"no_profiles_returned": (
1,
"ERROR",
"The FMP API returned earnings symbols but no company profiles for any of them.",
),
"profiles_missing_required_field:marketCap": (
1,
"ERROR",
"None of the returned profiles contain a marketCap or mktCap field; "
"the FMP response shape may have changed.",
),
"all_below_market_cap_floor": (
0,
"INFO",
"All candidates were below the minimum market cap floor.",
),
"no_input_candidates": (
0,
"INFO",
"No records in the input JSON met the minimum grade filter.",
),
}
def _get_candidates_mode_a(client: FMPClient, args) -> tuple[list[dict], Optional[str]]:
"""Get candidates from FMP earnings calendar (Mode A).
Returns:
``(candidates, reason)`` where ``reason`` is ``None`` when
``candidates`` is non-empty, else one of the keys in
``_ZERO_RESULT_REASONS``.
"""
# Calculate date range
to_date = datetime.now().strftime("%Y-%m-%d")
from_date = (datetime.now() - timedelta(days=args.lookback_days)).strftime("%Y-%m-%d")
print(f" Fetching earnings calendar: {from_date} to {to_date}")
earnings = client.get_earnings_calendar(from_date, to_date)
if not isinstance(earnings, list):
print(" ERROR: Earnings calendar fetch failed (no usable response body)")
return [], "calendar_fetch_failed"
if not earnings:
print(" WARNING: No earnings data returned")
return [], "no_earnings_rows"
print(f" Raw earnings events: {len(earnings)}")
# Get unique symbols (non-dict rows are ignored, never dereferenced).
symbols = list(
set(e.get("symbol", "") for e in earnings if isinstance(e, dict) and e.get("symbol"))
)
if not symbols:
return [], "no_earnings_rows"
# Fetch company profiles for market cap filtering
print(f" Fetching profiles for {len(symbols)} symbols...")
profiles = client.get_company_profiles(symbols)
if not profiles:
api_stats = client.get_api_stats()
if api_stats.get("budget_remaining") == 0 or api_stats.get("rate_limit_reached"):
return [], "profiles_budget_exhausted"
return [], "no_profiles_returned"
# Build candidates with market cap filter (gap filter deferred to Phase 2
# where actual price data is available for accurate gap calculation)
grade_map = {e.get("symbol"): e for e in earnings if isinstance(e, dict)}
candidates = []
any_usable_cap = False
for symbol in symbols:
earning = grade_map.get(symbol, {})
profile = profiles.get(symbol, {})
if isinstance(profile, dict) and (
_coerce_market_cap(profile.get("marketCap")) is not None
or _coerce_market_cap(profile.get("mktCap")) is not None
):
any_usable_cap = True
# Market cap filter (/stable returns marketCap; v3 returned mktCap)
market_cap = profile_market_cap(profile)
if market_cap < args.min_market_cap:
continue
timing = earning.get("time")
# Normalize timing. Anything that isn't a confirmed bmo/amc session
# (missing key, None/null from the provider, or an unrecognized
# string) becomes the canonical "unknown" -- never an empty string,
# so this matches the {"bmo", "amc", "unknown"} set that Mode B's
# validate_input_json requires (#352).
if timing in ("bmo", "Before Market Open"):
timing = "bmo"
elif timing in ("amc", "After Market Close"):
timing = "amc"
else:
timing = "unknown"
candidates.append(
{
"symbol": symbol,
"earnings_date": earning.get("date", ""),
"earnings_timing": timing,
"gap_pct": None, # Calculated from price data in Phase 2
"market_cap": market_cap,
}
)
print(f" Candidates after market cap filter: {len(candidates)}")
if candidates:
return candidates, None
if not any_usable_cap:
return [], "profiles_missing_required_field:marketCap"
return [], "all_below_market_cap_floor"
def _get_candidates_mode_b(args) -> tuple[list[dict], Optional[str]]:
"""Get candidates from earnings-trade-analyzer JSON (Mode B).
Returns:
``(candidates, reason)`` where ``reason`` is ``None`` when
``candidates`` is non-empty, else one of the keys in
``_ZERO_RESULT_REASONS``.
Raises:
SystemExit(1): On file not found, JSON parse error, or validation error.
"""
json_path = args.candidates_json
print(f" Loading: {json_path}")
if not os.path.exists(json_path):
print(f" ERROR: File not found: {json_path}", file=sys.stderr)
sys.exit(1)
with open(json_path) as f:
data = json.load(f)
# Validation errors are fatal in Mode B (bad input should not silently succeed)
validated = validate_input_json(data)
print(f" Validated records: {len(validated)}")
# Grade filter
grade_order = {"A": 0, "B": 1, "C": 2, "D": 3}
min_grade_rank = grade_order.get(args.min_grade, 1)
candidates = []
for record in validated:
grade = record.get("grade", "D")
grade_rank = grade_order.get(grade, 3)
if grade_rank <= min_grade_rank:
candidates.append(
{
"symbol": record["symbol"],
"earnings_date": record["earnings_date"],
"earnings_timing": record.get("earnings_timing", ""),
"gap_pct": record.get("gap_pct", 0),
"grade": grade,
}
)
print(f" After grade filter (>= {args.min_grade}): {len(candidates)}")
if candidates:
return candidates, None
return [], "no_input_candidates"
if __name__ == "__main__":
main()
scripts/tests/test_pead_screener.py
#!/usr/bin/env python3
"""
Tests for PEAD Screener modules.
Covers weekly candle conversion, breakout detection, liquidity scoring,
risk/reward calculation, composite scoring, pattern analysis, report
generation, Mode B validation, FMP client error handling, and edge cases.
"""
import json
import logging
import os
import sys
import tempfile
from datetime import date, timedelta
from unittest.mock import MagicMock, patch
import pytest
from calculators.breakout_calculator import calculate_breakout
from calculators.liquidity_calculator import calculate_liquidity
from calculators.risk_reward_calculator import calculate_risk_reward
from calculators.weekly_candle_calculator import (
analyze_weekly_pattern,
daily_to_weekly,
)
from fmp_client import ApiCallBudgetExceeded, FMPClient
from report_generator import generate_json_report, generate_markdown_report
from scorer import COMPONENT_WEIGHTS, calculate_composite_score
from screen_pead import (
_ZERO_RESULT_REASONS,
_get_candidates_mode_a,
_get_candidates_mode_b,
analyze_stock,
calculate_price_gap,
calculate_setup_quality,
main,
profile_market_cap,
validate_input_json,
)
def test_breakout_boundary_scores_and_volume_fail_closed_paths():
red = {"high": 100.0, "low": 95.0}
confirmed = [{"volume": 200}, {"volume": 100}, {"volume": 100}]
assert calculate_breakout(confirmed, {"high": 0}, 103)["score"] == 0.0
assert calculate_breakout(confirmed, red, 103)["score"] == 100.0
assert calculate_breakout(confirmed, red, 102)["score"] == 85.0
assert calculate_breakout([{"volume": 200}], red, 101)["score"] == 70.0
assert calculate_breakout([{"volume": 200}], red, 100.5)["score"] == 55.0
no_prior_volume = [{"volume": 200}, {"volume": 0}, {}]
assert calculate_breakout(no_prior_volume, red, 103)["volume_confirmation"] is False
def test_liquidity_boundary_scores_and_dollar_volume_fallback():
assert calculate_liquidity([], 100)["score"] == 15.0
assert calculate_liquidity([{"volume": 1_000_000, "close": 60}], 60)["score"] == 85.0
assert calculate_liquidity([{"volume": 1_000_000, "close": 30}], 30)["score"] == 70.0
fallback = calculate_liquidity([{"volume": 1_000_000, "close": 0}], 30)
assert fallback["adv20_dollar"] == 30_000_000
assert fallback["score"] == 70.0
two_of_three = calculate_liquidity([{"volume": 500_000, "close": 100}], 100)
assert two_of_three["score"] == 40.0
assert two_of_three["passes_all"] is False
def test_risk_reward_scores_every_contract_boundary():
red = {"high": 100.0, "low": 90.0}
assert calculate_risk_reward(100, red, target_multiplier=3.0)["score"] == 100.0
assert calculate_risk_reward(100, red, target_multiplier=2.5)["score"] == 85.0
assert calculate_risk_reward(100, red, target_multiplier=2.0)["score"] == 70.0
assert calculate_risk_reward(100, red, target_multiplier=1.5)["score"] == 50.0
assert calculate_risk_reward(100, red, target_multiplier=1.0)["score"] == 25.0
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_daily_prices(
n: int,
start_date: str = "2026-02-01",
base_price: float = 100.0,
daily_change: float = 0.0,
volume: int = 1_500_000,
green: bool = True,
) -> list[dict]:
"""Generate synthetic daily price data (most-recent-first).
Args:
n: Number of trading days
start_date: Earliest date in the series
base_price: Starting price
daily_change: Daily price change (additive)
volume: Daily volume
green: If True, close > open for each day
"""
start = date.fromisoformat(start_date)
prices = []
for i in range(n):
day = start + timedelta(days=i)
# Skip weekends
while day.weekday() >= 5:
day += timedelta(days=1)
p = base_price + daily_change * i
o = p if green else p + 0.5
c = p + 0.5 if green else p
prices.append(
{
"date": day.strftime("%Y-%m-%d"),
"open": round(o, 2),
"high": round(max(o, c) + 1.0, 2),
"low": round(min(o, c) - 1.0, 2),
"close": round(c, 2),
"volume": volume,
}
)
# Return most-recent-first
prices.reverse()
return prices
def _make_weekly_candle(
week_start: str,
year: int,
week: int,
open_: float,
high: float,
low: float,
close: float,
volume: int = 5_000_000,
partial: bool = False,
) -> dict:
"""Helper to build a weekly candle dict."""
return {
"week_start": week_start,
"year": year,
"week": week,
"open": open_,
"high": high,
"low": low,
"close": close,
"volume": volume,
"is_green": close >= open_,
"partial_week": partial,
"trading_days": 3 if partial else 5,
}
# ===========================================================================
# TestWeeklyCandleCalculator
# ===========================================================================
class TestWeeklyCandleCalculator:
"""Test daily_to_weekly conversion."""
def test_daily_to_weekly_basic(self):
"""10 trading days -> 2 weekly candles."""
# Create 10 days spanning 2 ISO weeks
# Mon 2026-02-02 to Fri 2026-02-13 (week 6 and 7)
prices = _make_daily_prices(10, start_date="2026-02-02", base_price=100.0)
weekly = daily_to_weekly(prices)
assert len(weekly) == 2
# Most-recent-first
assert weekly[0]["week"] >= weekly[1]["week"]
# Each candle should have OHLCV
for candle in weekly:
assert "open" in candle
assert "high" in candle
assert "low" in candle
assert "close" in candle
assert "volume" in candle
assert "is_green" in candle
assert "trading_days" in candle
def test_earnings_week_split(self):
"""Days before earnings_date are excluded from earnings week candle."""
# Week of 2026-02-02 (Mon) to 2026-02-06 (Fri)
# Earnings on Wednesday 2026-02-04
prices = _make_daily_prices(5, start_date="2026-02-02", base_price=100.0)
weekly = daily_to_weekly(prices, earnings_date="2026-02-04")
# The earnings week should only have Wed-Fri (3 days)
# Find the earnings week candle
earnings_week = None
for candle in weekly:
iso = date(2026, 2, 4).isocalendar()
if candle["year"] == iso[0] and candle["week"] == iso[1]:
earnings_week = candle
break
assert earnings_week is not None
assert earnings_week["trading_days"] == 3 # Wed, Thu, Fri
def test_partial_week(self):
"""Current incomplete week is marked partial_week=True."""
# Create prices where the latest day is a Wednesday (not Friday)
# 2026-02-18 is a Wednesday
prices = [
{
"date": "2026-02-18",
"open": 100,
"high": 102,
"low": 99,
"close": 101,
"volume": 1000000,
},
{
"date": "2026-02-17",
"open": 99,
"high": 101,
"low": 98,
"close": 100,
"volume": 1000000,
},
{
"date": "2026-02-16",
"open": 98,
"high": 100,
"low": 97,
"close": 99,
"volume": 1000000,
},
# Previous week (complete)
{
"date": "2026-02-13",
"open": 97,
"high": 99,
"low": 96,
"close": 98,
"volume": 1000000,
},
{
"date": "2026-02-12",
"open": 96,
"high": 98,
"low": 95,
"close": 97,
"volume": 1000000,
},
{
"date": "2026-02-11",
"open": 95,
"high": 97,
"low": 94,
"close": 96,
"volume": 1000000,
},
{
"date": "2026-02-10",
"open": 94,
"high": 96,
"low": 93,
"close": 95,
"volume": 1000000,
},
{
"date": "2026-02-09",
"open": 93,
"high": 95,
"low": 92,
"close": 94,
"volume": 1000000,
},
]
weekly = daily_to_weekly(prices)
# Most recent week should be partial
assert weekly[0]["partial_week"] is True
# Previous week should not be partial
assert weekly[1]["partial_week"] is False
def test_iso_week_monday_start(self):
"""Verify weeks are grouped by ISO week (Monday start)."""
# 2026-02-02 is a Monday
prices = _make_daily_prices(5, start_date="2026-02-02", base_price=100.0)
weekly = daily_to_weekly(prices)
# Should produce exactly 1 week (Mon-Fri)
assert len(weekly) == 1
# week_start should be a Monday
ws = date.fromisoformat(weekly[0]["week_start"])
assert ws.weekday() == 0 # 0 = Monday
def test_empty_input(self):
"""Empty daily prices returns empty weekly candles."""
assert daily_to_weekly([]) == []
# ===========================================================================
# TestBreakoutCalculator
# ===========================================================================
class TestBreakoutCalculator:
def test_breakout_detected(self):
"""Current price above red candle high -> breakout detected."""
weekly = [
_make_weekly_candle(
"2026-02-16", 2026, 8, 100, 105, 99, 104, volume=6_000_000
), # Green, current
_make_weekly_candle("2026-02-09", 2026, 7, 102, 103, 97, 98, volume=4_000_000), # Red
_make_weekly_candle("2026-02-02", 2026, 6, 95, 102, 94, 101, volume=5_000_000), # Green
]
red_candle = {
"high": 103,
"low": 97,
"open": 102,
"close": 98,
"week_start": "2026-02-09",
"week_index": 1,
}
result = calculate_breakout(weekly, red_candle, current_price=104)
assert result["is_breakout"] is True
assert result["breakout_pct"] > 0
def test_no_breakout(self):
"""Current price below red candle high -> no breakout."""
weekly = [
_make_weekly_candle("2026-02-16", 2026, 8, 100, 102, 99, 101, volume=4_000_000),
]
red_candle = {
"high": 103,
"low": 97,
"open": 102,
"close": 98,
"week_start": "2026-02-09",
"week_index": 1,
}
result = calculate_breakout(weekly, red_candle, current_price=101)
assert result["is_breakout"] is False
assert result["score"] == 0.0
def test_volume_confirmation(self):
"""Breakout with volume above 4-week average -> volume_confirmation=True."""
weekly = [
_make_weekly_candle(
"2026-02-16", 2026, 8, 100, 107, 99, 106, volume=8_000_000
), # High vol breakout
_make_weekly_candle("2026-02-09", 2026, 7, 102, 103, 97, 98, volume=3_000_000),
_make_weekly_candle("2026-02-02", 2026, 6, 95, 102, 94, 101, volume=3_000_000),
_make_weekly_candle("2026-01-26", 2026, 5, 90, 96, 89, 95, volume=3_000_000),
_make_weekly_candle("2026-01-19", 2026, 4, 88, 91, 87, 90, volume=3_000_000),
]
red_candle = {
"high": 103,
"low": 97,
"open": 102,
"close": 98,
"week_start": "2026-02-09",
"week_index": 1,
}
result = calculate_breakout(weekly, red_candle, current_price=106)
assert result["volume_confirmation"] is True
assert result["score"] >= 85
def test_no_red_candle(self):
"""No red candle -> score 0."""
weekly = [_make_weekly_candle("2026-02-16", 2026, 8, 100, 105, 99, 104)]
result = calculate_breakout(weekly, None, current_price=104)
assert result["score"] == 0.0
# ===========================================================================
# TestLiquidityCalculator
# ===========================================================================
class TestLiquidityCalculator:
def test_passes_all(self):
"""High liquidity stock passes all thresholds."""
prices = _make_daily_prices(20, base_price=150.0, volume=2_000_000)
result = calculate_liquidity(prices, current_price=150.0)
assert result["passes_all"] is True
assert result["score"] >= 70
def test_fails_one(self):
"""Moderate price stock fails only ADV20 threshold (price + volume pass)."""
# Price=$15 passes ($10+), Volume=2M passes (1M+),
# but ADV20=$15*2M=$30M barely above $25M — use lower volume to fail ADV20
prices = _make_daily_prices(20, base_price=15.0, volume=500_000)
calculate_liquidity(prices, current_price=15.0)
# ADV20 = $15 * 500K = $7.5M < $25M (fail), volume 500K < 1M (fail), price $15 >= $10 (pass)
# That's 1 of 3 pass -> score 15. Let's use params where exactly 2 pass.
# Price=$15 (pass), Volume=2M (pass), ADV20=$15*2M=$30M (pass) -> all pass
# Instead: price=$8 (fail), volume=5M (pass), ADV20=$8*5M=$40M (pass) -> 2 pass
prices2 = _make_daily_prices(20, base_price=8.0, volume=5_000_000)
result2 = calculate_liquidity(prices2, current_price=8.0)
assert result2["passes_all"] is False
assert result2["score"] == 40 # 2 of 3 pass (volume + ADV20, price fails)
def test_fails_all(self):
"""Low price, low volume -> fails all."""
prices = _make_daily_prices(20, base_price=3.0, volume=50_000)
result = calculate_liquidity(prices, current_price=3.0)
assert result["passes_all"] is False
assert result["score"] == 15
def test_high_adv20(self):
"""Very high dollar volume -> score 100."""
prices = _make_daily_prices(20, base_price=200.0, volume=5_000_000)
result = calculate_liquidity(prices, current_price=200.0)
assert result["passes_all"] is True
assert result["adv20_dollar"] > 100_000_000
assert result["score"] == 100
# ===========================================================================
# TestRiskRewardCalculator
# ===========================================================================
class TestRiskRewardCalculator:
def test_good_rr(self):
"""R:R >= 2.0 -> score 70."""
red_candle = {"high": 100, "low": 95, "open": 100, "close": 96}
result = calculate_risk_reward(
current_price=101, red_candle=red_candle, target_multiplier=2.0
)
assert result["risk_reward_ratio"] == 2.0
assert result["score"] == 70
def test_poor_rr(self):
"""Small risk distance -> high R:R but very small absolute risk."""
red_candle = {"high": 100, "low": 99.5, "open": 100, "close": 99.6}
result = calculate_risk_reward(
current_price=100.1, red_candle=red_candle, target_multiplier=2.0
)
assert result["risk_reward_ratio"] == 2.0
assert result["score"] == 70
def test_excellent_rr(self):
"""R:R >= 3.0 -> score 100."""
red_candle = {"high": 100, "low": 95, "open": 100, "close": 96}
result = calculate_risk_reward(
current_price=101, red_candle=red_candle, target_multiplier=3.0
)
assert result["risk_reward_ratio"] == 3.0
assert result["score"] == 100
def test_no_red_candle(self):
"""No red candle -> default score 25."""
result = calculate_risk_reward(current_price=100, red_candle=None)
assert result["score"] == 25
def test_stop_above_entry(self):
"""Stop >= entry (invalid) -> default score 25."""
red_candle = {"high": 100, "low": 105, "open": 100, "close": 104}
result = calculate_risk_reward(current_price=100, red_candle=red_candle)
assert result["score"] == 25
# ===========================================================================
# TestScorer
# ===========================================================================
class TestScorer:
def test_all_high_strong_setup(self):
"""All scores 100 -> Strong Setup."""
result = calculate_composite_score(100, 100, 100, 100)
assert result["composite_score"] == 100.0
assert result["rating"] == "Strong Setup"
def test_all_low_weak(self):
"""All scores 0 -> Weak."""
result = calculate_composite_score(0, 0, 0, 0)
assert result["composite_score"] == 0.0
assert result["rating"] == "Weak"
def test_weights_sum_to_1(self):
"""Verify component weights sum to 1.0."""
total = sum(COMPONENT_WEIGHTS.values())
assert abs(total - 1.0) < 0.001
def test_mixed_scores_good(self):
"""Mixed scores in the Good range."""
# 80*0.30 + 75*0.25 + 70*0.25 + 65*0.20 = 24 + 18.75 + 17.5 + 13 = 73.25
result = calculate_composite_score(80, 75, 70, 65)
assert 70 <= result["composite_score"] < 85
assert result["rating"] == "Good Setup"
def test_component_breakdown_present(self):
"""Result includes component breakdown."""
result = calculate_composite_score(80, 70, 60, 50)
assert "component_breakdown" in result
assert len(result["component_breakdown"]) == 4
def test_weakest_strongest(self):
"""Weakest and strongest components identified correctly."""
result = calculate_composite_score(90, 80, 70, 60)
assert result["weakest_component"] == "Risk/Reward"
assert result["weakest_score"] == 60
assert result["strongest_component"] == "Setup Quality"
assert result["strongest_score"] == 90
# ===========================================================================
# TestWeeklyPattern
# ===========================================================================
class TestWeeklyPattern:
"""Test analyze_weekly_pattern stage classification."""
def _build_candles_for_stage(self, stage: str) -> tuple:
"""Build weekly candles that produce the given stage."""
# Earnings on 2026-02-02 (Monday, ISO week 6)
earnings_date = "2026-02-02"
if stage == "MONITORING":
# All green candles after earnings, no red candle
candles = [
_make_weekly_candle("2026-02-09", 2026, 7, 105, 110, 104, 109), # Green
_make_weekly_candle(
"2026-02-02", 2026, 6, 100, 106, 99, 105
), # Green (earnings week)
]
elif stage == "SIGNAL_READY":
# Red candle exists but no breakout
candles = [
_make_weekly_candle(
"2026-02-16", 2026, 8, 106, 108, 104, 105
), # Green, below red high
_make_weekly_candle(
"2026-02-09", 2026, 7, 109, 110, 104, 106
), # Red (close < open)
_make_weekly_candle(
"2026-02-02", 2026, 6, 100, 110, 99, 109
), # Green (earnings week)
]
elif stage == "BREAKOUT":
# Red candle exists and current candle breaks above it
candles = [
_make_weekly_candle(
"2026-02-16", 2026, 8, 108, 115, 107, 114
), # Green, above red high=110
_make_weekly_candle("2026-02-09", 2026, 7, 109, 110, 104, 106), # Red
_make_weekly_candle(
"2026-02-02", 2026, 6, 100, 110, 99, 109
), # Green (earnings week)
]
elif stage == "EXPIRED":
# More than 5 weeks since earnings
candles = [
_make_weekly_candle("2026-03-16", 2026, 12, 105, 108, 104, 107),
_make_weekly_candle("2026-03-09", 2026, 11, 104, 107, 103, 106),
_make_weekly_candle("2026-03-02", 2026, 10, 103, 106, 102, 105),
_make_weekly_candle("2026-02-23", 2026, 9, 102, 105, 101, 104),
_make_weekly_candle("2026-02-16", 2026, 8, 101, 104, 100, 103),
_make_weekly_candle("2026-02-09", 2026, 7, 100, 103, 99, 102),
_make_weekly_candle("2026-02-02", 2026, 6, 95, 101, 94, 100), # Earnings week
]
else:
candles = []
return candles, earnings_date
def test_monitoring_stage(self):
candles, earnings_date = self._build_candles_for_stage("MONITORING")
result = analyze_weekly_pattern(candles, earnings_date)
assert result["stage"] == "MONITORING"
assert result["red_candle"] is None
def test_signal_ready_stage(self):
candles, earnings_date = self._build_candles_for_stage("SIGNAL_READY")
result = analyze_weekly_pattern(candles, earnings_date)
assert result["stage"] == "SIGNAL_READY"
assert result["red_candle"] is not None
assert result["is_breakout"] is False
def test_breakout_stage(self):
candles, earnings_date = self._build_candles_for_stage("BREAKOUT")
result = analyze_weekly_pattern(candles, earnings_date)
assert result["stage"] == "BREAKOUT"
assert result["is_breakout"] is True
assert result["breakout_pct"] > 0
def test_expired_stage(self):
candles, earnings_date = self._build_candles_for_stage("EXPIRED")
result = analyze_weekly_pattern(candles, earnings_date, watch_weeks=5)
assert result["stage"] == "EXPIRED"
# ===========================================================================
# TestReportGenerator
# ===========================================================================
class TestReportGenerator:
def _make_result(self, symbol="TEST", stage="BREAKOUT", score=82.5):
return {
"symbol": symbol,
"stage": stage,
"earnings_date": "2026-02-15",
"earnings_timing": "amc",
"gap_pct": 6.3,
"weeks_since_earnings": 2,
"red_candle": {
"high": 195.0,
"low": 188.0,
"week_start": "2026-02-17",
"open": 194.0,
"close": 189.0,
"week_index": 1,
},
"current_price": 197.5,
"breakout_pct": 1.28,
"entry_price": 197.5,
"stop_price": 188.0,
"target_price": 216.5,
"risk_pct": 4.81,
"risk_reward_ratio": 2.0,
"adv20_dollar": 45_000_000,
"composite_score": score,
"rating": "Good Setup" if score >= 70 else "Developing",
"guidance": "Solid PEAD setup, standard position size",
"components": {
"setup_quality": {
"label": "Setup Quality",
"score": 90,
"weight": 0.30,
"weighted": 27.0,
},
"breakout_strength": {
"label": "Breakout Strength",
"score": 70,
"weight": 0.25,
"weighted": 17.5,
},
"liquidity": {"label": "Liquidity", "score": 85, "weight": 0.25, "weighted": 21.25},
"risk_reward": {
"label": "Risk/Reward",
"score": 70,
"weight": 0.20,
"weighted": 14.0,
},
},
}
def test_json_structure(self):
"""JSON report has correct structure."""
results = [self._make_result("AAPL", "BREAKOUT", 85)]
metadata = {
"generated_at": "2026-02-21 10:00:00",
"lookback_days": 14,
"watch_weeks": 5,
"mode": "A",
"api_stats": {"api_calls_made": 50, "budget_remaining": 150},
}
with tempfile.TemporaryDirectory() as tmpdir:
json_file = os.path.join(tmpdir, "test.json")
generate_json_report(results, metadata, json_file)
with open(json_file) as f:
data = json.load(f)
assert "metadata" in data
assert "results" in data
assert "summary" in data
assert data["summary"]["breakout"] == 1
def test_markdown_generation(self):
"""Markdown report generates without errors."""
results = [
self._make_result("AAPL", "BREAKOUT", 85),
self._make_result("MSFT", "SIGNAL_READY", 65),
self._make_result("GOOG", "MONITORING", 40),
]
metadata = {
"generated_at": "2026-02-21 10:00:00",
"lookback_days": 14,
"watch_weeks": 5,
"mode": "A",
"api_stats": {"api_calls_made": 50, "budget_remaining": 150},
}
with tempfile.TemporaryDirectory() as tmpdir:
md_file = os.path.join(tmpdir, "test.md")
generate_markdown_report(results, metadata, md_file)
with open(md_file) as f:
content = f.read()
assert "PEAD Screener Report" in content
assert "BREAKOUT" in content
assert "SIGNAL_READY" in content
assert "MONITORING" in content
assert "AAPL" in content
assert "MSFT" in content
assert "GOOG" in content
def test_markdown_shows_timing_unknown_row_for_mode_a(self):
results = [self._make_result("AAPL", "BREAKOUT", 85)]
metadata = {
"generated_at": "2026-02-21 10:00:00",
"lookback_days": 14,
"watch_weeks": 5,
"mode": "A",
"api_stats": {"api_calls_made": 50, "budget_remaining": 150},
"timing_unknown_count": 2,
"timing_candidates_total": 6,
}
with tempfile.TemporaryDirectory() as tmpdir:
md_file = os.path.join(tmpdir, "test.md")
generate_markdown_report(results, metadata, md_file)
with open(md_file) as f:
content = f.read()
assert "| Timing unknown | 2 of 6 |" in content
def test_markdown_shows_timing_unknown_as_na_for_mode_b(self):
results = [self._make_result("AAPL", "BREAKOUT", 85)]
metadata = {
"generated_at": "2026-02-21 10:00:00",
"lookback_days": None,
"watch_weeks": 5,
"mode": "B",
"api_stats": {"api_calls_made": 50, "budget_remaining": 150},
"timing_unknown_count": None,
"timing_candidates_total": None,
}
with tempfile.TemporaryDirectory() as tmpdir:
md_file = os.path.join(tmpdir, "test.md")
generate_markdown_report(results, metadata, md_file)
with open(md_file) as f:
content = f.read()
assert "| Timing unknown | n/a |" in content
def test_stage_grouping(self):
"""Results are grouped by stage in the report."""
results = [
self._make_result("A", "MONITORING", 40),
self._make_result("B", "BREAKOUT", 85),
self._make_result("C", "SIGNAL_READY", 65),
]
metadata = {
"generated_at": "2026-02-21",
"mode": "A",
"watch_weeks": 5,
"lookback_days": 14,
"api_stats": {},
}
with tempfile.TemporaryDirectory() as tmpdir:
json_file = os.path.join(tmpdir, "test.json")
generate_json_report(results, metadata, json_file)
with open(json_file) as f:
data = json.load(f)
# First result should be BREAKOUT (highest priority)
assert data["results"][0]["stage"] == "BREAKOUT"
# ===========================================================================
# TestValidateInputJson (Mode B Failure Cases)
# ===========================================================================
class TestValidateInputJson:
def test_schema_version_mismatch(self):
"""schema_version '2.0' -> ValueError."""
data = {
"schema_version": "2.0",
"results": [
{
"symbol": "AAPL",
"earnings_date": "2026-02-15",
"earnings_timing": "amc",
"gap_pct": 5.0,
"grade": "A",
}
],
}
try:
validate_input_json(data)
raise AssertionError("Should have raised ValueError")
except ValueError as e:
assert "Schema version mismatch" in str(e)
assert "2.0" in str(e)
def test_missing_required_field(self, caplog):
"""Result missing 'symbol' -> skip + warning log."""
data = {
"schema_version": "1.0",
"results": [
{
"symbol": "AAPL",
"earnings_date": "2026-02-15",
"earnings_timing": "amc",
"gap_pct": 5.0,
"grade": "A",
},
{
"earnings_date": "2026-02-16", # Missing 'symbol'
"earnings_timing": "bmo",
"gap_pct": 3.0,
"grade": "B",
},
],
}
with caplog.at_level(logging.WARNING):
validated = validate_input_json(data)
assert len(validated) == 1
assert validated[0]["symbol"] == "AAPL"
# Check warning was logged
assert any("missing required fields" in r.message.lower() for r in caplog.records)
def test_valid_input(self):
"""Proper schema '1.0' with all fields -> passes."""
data = {
"schema_version": "1.0",
"results": [
{
"symbol": "AAPL",
"earnings_date": "2026-02-15",
"earnings_timing": "amc",
"gap_pct": 5.0,
"grade": "A",
},
{
"symbol": "MSFT",
"earnings_date": "2026-02-16",
"earnings_timing": "bmo",
"gap_pct": 3.5,
"grade": "B",
},
],
}
validated = validate_input_json(data)
assert len(validated) == 2
def test_all_records_invalid(self):
"""All records missing fields -> ValueError (empty results)."""
data = {
"schema_version": "1.0",
"results": [
{"earnings_date": "2026-02-15"}, # Missing symbol, timing, gap, grade
{"gap_pct": 5.0}, # Missing symbol, date, timing, grade
],
}
try:
validate_input_json(data)
raise AssertionError("Should have raised ValueError")
except ValueError as e:
assert "All" in str(e) and "records failed" in str(e)
def test_invalid_timing_skipped(self, caplog):
"""Invalid earnings_timing value -> skip + warning."""
data = {
"schema_version": "1.0",
"results": [
{
"symbol": "AAPL",
"earnings_date": "2026-02-15",
"earnings_timing": "amc",
"gap_pct": 5.0,
"grade": "A",
},
{
"symbol": "MSFT",
"earnings_date": "2026-02-16",
"earnings_timing": "invalid_timing",
"gap_pct": 3.0,
"grade": "B",
},
],
}
with caplog.at_level(logging.WARNING):
validated = validate_input_json(data)
assert len(validated) == 1
assert validated[0]["symbol"] == "AAPL"
def test_invalid_grade_skipped(self, caplog):
"""Invalid grade value -> skip + warning."""
data = {
"schema_version": "1.0",
"results": [
{
"symbol": "AAPL",
"earnings_date": "2026-02-15",
"earnings_timing": "amc",
"gap_pct": 5.0,
"grade": "A",
},
{
"symbol": "MSFT",
"earnings_date": "2026-02-16",
"earnings_timing": "bmo",
"gap_pct": 3.0,
"grade": "Z",
},
],
}
with caplog.at_level(logging.WARNING):
validated = validate_input_json(data)
assert len(validated) == 1
def test_gap_pct_string_skipped(self, caplog):
"""gap_pct as string -> skip + warning."""
data = {
"schema_version": "1.0",
"results": [
{
"symbol": "AAPL",
"earnings_date": "2026-02-15",
"earnings_timing": "amc",
"gap_pct": "not_a_number",
"grade": "A",
},
{
"symbol": "MSFT",
"earnings_date": "2026-02-16",
"earnings_timing": "bmo",
"gap_pct": 3.0,
"grade": "B",
},
],
}
with caplog.at_level(logging.WARNING):
validated = validate_input_json(data)
assert len(validated) == 1
assert validated[0]["symbol"] == "MSFT"
def test_empty_symbol_skipped(self, caplog):
"""Empty symbol string -> skip + warning."""
data = {
"schema_version": "1.0",
"results": [
{
"symbol": "",
"earnings_date": "2026-02-15",
"earnings_timing": "amc",
"gap_pct": 5.0,
"grade": "A",
},
{
"symbol": "MSFT",
"earnings_date": "2026-02-16",
"earnings_timing": "bmo",
"gap_pct": 3.0,
"grade": "B",
},
],
}
with caplog.at_level(logging.WARNING):
validated = validate_input_json(data)
assert len(validated) == 1
assert validated[0]["symbol"] == "MSFT"
# ===========================================================================
# TestCalculatePriceGap (Fix #1: actual price gap in Mode A)
# ===========================================================================
class TestCalculatePriceGap:
"""Test actual price gap calculation from daily OHLCV data."""
def test_bmo_gap(self):
"""BMO: gap = open[earnings_date] / close[prev_day] - 1"""
prices = [
{
"date": "2026-02-17",
"open": 110.0,
"high": 112.0,
"low": 109.0,
"close": 111.0,
"volume": 1000000,
},
{
"date": "2026-02-16",
"open": 106.0,
"high": 111.0,
"low": 105.0,
"close": 110.0,
"volume": 3000000,
},
{
"date": "2026-02-13",
"open": 99.0,
"high": 101.0,
"low": 98.0,
"close": 100.0,
"volume": 1000000,
},
]
# BMO on 2026-02-16: gap = 106.0 / 100.0 - 1 = 6.0%
gap = calculate_price_gap(prices, "2026-02-16", "bmo")
assert gap == 6.0
def test_amc_gap(self):
"""AMC: gap = open[next_day] / close[earnings_date] - 1"""
prices = [
{
"date": "2026-02-17",
"open": 108.0,
"high": 110.0,
"low": 107.0,
"close": 109.0,
"volume": 2000000,
},
{
"date": "2026-02-16",
"open": 99.0,
"high": 101.0,
"low": 98.0,
"close": 100.0,
"volume": 3000000,
},
{
"date": "2026-02-13",
"open": 98.0,
"high": 100.0,
"low": 97.0,
"close": 99.0,
"volume": 1000000,
},
]
# AMC on 2026-02-16: gap = 108.0 / 100.0 - 1 = 8.0%
gap = calculate_price_gap(prices, "2026-02-16", "amc")
assert gap == 8.0
def test_unknown_timing_uses_amc(self):
"""Unknown timing falls back to AMC logic."""
prices = [
{
"date": "2026-02-17",
"open": 105.0,
"high": 107.0,
"low": 104.0,
"close": 106.0,
"volume": 1000000,
},
{
"date": "2026-02-16",
"open": 99.0,
"high": 101.0,
"low": 98.0,
"close": 100.0,
"volume": 3000000,
},
]
gap_amc = calculate_price_gap(prices, "2026-02-16", "amc")
gap_unknown = calculate_price_gap(prices, "2026-02-16", "")
assert gap_amc == gap_unknown
def test_missing_earnings_date(self):
"""Earnings date not in data -> 0.0."""
prices = [
{"date": "2026-02-17", "open": 100.0, "close": 101.0},
]
gap = calculate_price_gap(prices, "2099-12-31", "bmo")
assert gap == 0.0
def test_no_prev_day_bmo(self):
"""BMO with no previous day -> 0.0."""
prices = [
{"date": "2026-02-16", "open": 106.0, "close": 107.0},
]
gap = calculate_price_gap(prices, "2026-02-16", "bmo")
assert gap == 0.0
# ===========================================================================
# TestFMPClient (Error Handling)
# ===========================================================================
class TestFMPClient:
@patch("fmp_client.requests.Session")
def test_api_429(self, mock_session_class):
"""Mock 429 -> rate_limit_reached, returns None."""
mock_session = MagicMock()
mock_response = MagicMock()
mock_response.status_code = 429
mock_response.text = "Rate limit exceeded"
mock_session.get.return_value = mock_response
mock_session_class.return_value = mock_session
client = FMPClient(api_key="test_key", max_api_calls=200)
client.session = mock_session
client.max_retries = 0 # Don't retry for test speed
result = client.get_earnings_calendar("2026-02-01", "2026-02-15")
assert result is None
assert client.rate_limit_reached is True
@patch("fmp_client.requests.Session")
def test_api_timeout(self, mock_session_class):
"""Mock Timeout -> returns None."""
import requests as req
mock_session = MagicMock()
mock_session.get.side_effect = req.exceptions.Timeout("Connection timed out")
mock_session_class.return_value = mock_session
client = FMPClient(api_key="test_key", max_api_calls=200)
client.session = mock_session
result = client.get_historical_prices("AAPL", days=90)
assert result is None
def test_budget_exceeded(self):
"""Raises ApiCallBudgetExceeded when budget exhausted."""
client = FMPClient(api_key="test_key", max_api_calls=0)
try:
client._rate_limited_get("https://example.com/api/v3/test")
raise AssertionError("Should have raised ApiCallBudgetExceeded")
except ApiCallBudgetExceeded as e:
assert "budget exhausted" in str(e).lower()
class TestFMPHistoricalNormalizer:
"""Cover stable/historical-price-eod/full flat-list normalization (Issue #64)."""
@staticmethod
def _mock_response(status_code, json_payload):
resp = MagicMock()
resp.status_code = status_code
resp.json.return_value = json_payload
resp.text = ""
return resp
@staticmethod
def _make_client(mock_session):
client = FMPClient(api_key="test_key", max_api_calls=200)
client.session = mock_session
client.max_retries = 0
return client
@patch("fmp_client.requests.Session")
def test_eod_flat_list_truncated_to_days(self, mock_session_class):
"""Contract: returned historical is truncated to `days` rows.
The new EOD endpoint ignores `timeseries` and returns full history.
The normalizer must truncate so callers receive at most N rows,
preserving the legacy v3 `timeseries=N` contract.
"""
mock_session = MagicMock()
mock_session.get.return_value = self._mock_response(
200,
[
{
"symbol": "SPY",
"date": f"2026-04-{30 - i:02d}",
"open": 500.0,
"high": 502.0,
"low": 499.0,
"close": 500.0 + i,
"volume": 1_000_000,
}
for i in range(5)
],
)
mock_session_class.return_value = mock_session
client = self._make_client(mock_session)
result = client.get_historical_prices("SPY", days=2)
assert result is not None
# API returned 5 rows but we requested days=2; normalizer must truncate
assert len(result["historical"]) == 2, (
f"expected truncation to 2 rows, got {len(result['historical'])}"
)
# Most recent first preserved
assert result["historical"][0]["date"] == "2026-04-30"
assert result["historical"][1]["date"] == "2026-04-29"
@patch("fmp_client.requests.Session")
def test_eod_flat_list_normalized(self, mock_session_class):
"""New stable EOD flat list -> v3-compat dict with historical[]."""
mock_session = MagicMock()
mock_session.get.return_value = self._mock_response(
200,
[
{
"symbol": "SPY",
"date": "2026-04-29",
"open": 500.0,
"high": 502.0,
"low": 499.0,
"close": 501.0,
"volume": 1_000_000,
},
{
"symbol": "SPY",
"date": "2026-04-28",
"open": 498.0,
"high": 501.0,
"low": 497.0,
"close": 500.0,
"volume": 1_100_000,
},
],
)
mock_session_class.return_value = mock_session
client = self._make_client(mock_session)
result = client.get_historical_prices("SPY", days=2)
assert result is not None
assert result["symbol"] == "SPY"
assert len(result["historical"]) == 2
assert result["historical"][0]["date"] == "2026-04-29"
assert result["historical"][0]["close"] == 501.0
assert "symbol" not in result["historical"][0], (
"row-level symbol should be stripped to mirror v3 shape"
)
@patch("fmp_client.requests.Session")
def test_empty_list_falls_back_via_falsy_path(self, mock_session_class):
"""Empty list response is caught by `if not data: continue` before normalizer."""
mock_session = MagicMock()
mock_session.get.return_value = self._mock_response(200, [])
mock_session_class.return_value = mock_session
client = self._make_client(mock_session)
result = client.get_historical_prices("SPY", days=2)
# Both stable and v3 fallback see empty/None -> final result None
assert result is None
@patch("fmp_client.requests.Session")
def test_eod_symbol_mismatch_rejected(self, mock_session_class):
"""List with no matching symbol -> normalizer returns None, fallback exhausted."""
mock_session = MagicMock()
mock_session.get.return_value = self._mock_response(
200,
[{"symbol": "QQQ", "date": "2026-04-29", "open": 1.0, "close": 1.0}],
)
mock_session_class.return_value = mock_session
client = self._make_client(mock_session)
result = client.get_historical_prices("SPY", days=2)
assert result is None
@patch("fmp_client.requests.Session")
def test_eod_row_without_symbol_field(self, mock_session_class):
"""Single-symbol endpoint may omit per-row 'symbol' -> treat as requested symbol."""
mock_session = MagicMock()
mock_session.get.return_value = self._mock_response(
200,
[
{"date": "2026-04-29", "open": 500.0, "close": 501.0},
{"date": "2026-04-28", "open": 498.0, "close": 500.0},
],
)
mock_session_class.return_value = mock_session
client = self._make_client(mock_session)
result = client.get_historical_prices("SPY", days=2)
assert result is not None
assert result["symbol"] == "SPY"
assert len(result["historical"]) == 2
assert result["historical"][0]["close"] == 501.0
@patch("fmp_client.requests.Session")
def test_eod_index_symbol_normalized(self, mock_session_class):
"""Dot/dash normalization (BRK.B vs BRK-B style) works."""
mock_session = MagicMock()
mock_session.get.return_value = self._mock_response(
200,
[{"symbol": "BRK.B", "date": "2026-04-29", "open": 400.0, "close": 401.0}],
)
mock_session_class.return_value = mock_session
client = self._make_client(mock_session)
result = client.get_historical_prices("BRK-B", days=1)
assert result is not None
assert result["historical"][0]["close"] == 401.0
@patch("fmp_client.requests.Session")
def test_legacy_v3_dict_passthrough(self, mock_session_class):
"""v3 fallback dict shape passes normalizer untouched."""
mock_session = MagicMock()
mock_session.get.return_value = self._mock_response(
200,
{"symbol": "SPY", "historical": [{"date": "2026-04-29", "close": 501.0}]},
)
mock_session_class.return_value = mock_session
client = self._make_client(mock_session)
result = client.get_historical_prices("SPY", days=1)
assert result is not None
assert result["historical"][0]["close"] == 501.0
@patch("fmp_client.requests.Session")
def test_legacy_historicalStockList_still_works(self, mock_session_class):
"""historicalStockList batch shape still handled by existing branch."""
mock_session = MagicMock()
mock_session.get.return_value = self._mock_response(
200,
{
"historicalStockList": [
{"symbol": "SPY", "historical": [{"date": "2026-04-29", "close": 501.0}]},
{"symbol": "QQQ", "historical": [{"date": "2026-04-29", "close": 400.0}]},
]
},
)
mock_session_class.return_value = mock_session
client = self._make_client(mock_session)
result = client.get_historical_prices("SPY", days=1)
assert result is not None
assert result["symbol"] == "SPY"
assert result["historical"][0]["close"] == 501.0
@patch("fmp_client.requests.Session")
def test_url_uses_eod_endpoint(self, mock_session_class):
"""Regression: stable URL must be /historical-price-eod/full, not /historical-price-full."""
mock_session = MagicMock()
mock_session.get.return_value = self._mock_response(
200,
[{"symbol": "SPY", "date": "2026-04-29", "close": 1.0}],
)
mock_session_class.return_value = mock_session
client = self._make_client(mock_session)
client.get_historical_prices("SPY", days=1)
# First call should hit the new EOD URL with from/to params (not timeseries)
first_call = mock_session.get.call_args_list[0]
url = first_call[0][0]
params = first_call[1]["params"]
assert "historical-price-eod/full" in url
assert "historical-price-full" not in url.replace("historical-price-eod/full", "")
assert params.get("symbol") == "SPY"
assert "from" in params and "to" in params
assert "timeseries" not in params, (
"timeseries must be converted to from/to since stable EOD ignores timeseries"
)
# ===========================================================================
# TestPartialWeekBoundary
# ===========================================================================
class TestPartialWeekBoundary:
def test_friday_only_week(self):
"""Single Friday -> valid partial weekly candle."""
# 2026-02-20 is a Friday
prices = [
{
"date": "2026-02-20",
"open": 100,
"high": 102,
"low": 99,
"close": 101,
"volume": 1000000,
},
]
weekly = daily_to_weekly(prices)
assert len(weekly) == 1
# It's the only day in the week, and it's Friday (day 5 in ISO)
# Since isocalendar day_of_week is 5 (Friday), it should NOT be partial
assert weekly[0]["trading_days"] == 1
assert weekly[0]["open"] == 100
assert weekly[0]["close"] == 101
def test_monday_earnings_bmo(self):
"""Monday BMO earnings -> earnings week starts Monday (full week)."""
# 2026-02-16 is a Monday
prices = _make_daily_prices(5, start_date="2026-02-16", base_price=100.0)
weekly = daily_to_weekly(prices, earnings_date="2026-02-16")
# Earnings on Monday BMO -> include Monday and all subsequent days
assert len(weekly) == 1
assert weekly[0]["trading_days"] == 5 # Full week Mon-Fri
# ===========================================================================
# TestSetupQuality
# ===========================================================================
class TestSetupQuality:
"""Test calculate_setup_quality scoring."""
def test_large_gap_breakout(self):
"""10%+ gap with BREAKOUT stage -> high score."""
pattern = {"stage": "BREAKOUT", "weeks_since_earnings": 2, "red_candle": None}
score = calculate_setup_quality(10.0, pattern)
assert score == 100 # 50 (gap) + 50 (breakout)
def test_small_gap_monitoring(self):
"""3% gap with MONITORING early -> moderate score."""
pattern = {"stage": "MONITORING", "weeks_since_earnings": 1, "red_candle": None}
score = calculate_setup_quality(3.0, pattern)
assert score == 45 # 20 (gap) + 25 (monitoring early)
def test_expired_stage(self):
"""EXPIRED stage -> only gap points."""
pattern = {"stage": "EXPIRED", "weeks_since_earnings": 6, "red_candle": None}
score = calculate_setup_quality(5.0, pattern)
assert score == 30 # 30 (gap) + 0 (expired)
# ===========================================================================
# TestAnalyzeStock (Integration)
# ===========================================================================
class TestAnalyzeStock:
"""Integration test for analyze_stock."""
def test_basic_analysis(self):
"""analyze_stock returns expected structure."""
# Create daily prices spanning 3+ weeks
prices = _make_daily_prices(30, start_date="2026-01-19", base_price=100.0, daily_change=0.5)
result = analyze_stock(
symbol="TEST",
daily_prices=prices,
earnings_date="2026-01-19",
earnings_timing="bmo",
gap_pct=5.0,
current_price=115.0,
watch_weeks=5,
)
assert result is not None
assert result["symbol"] == "TEST"
assert "stage" in result
assert "composite_score" in result
assert "rating" in result
assert "components" in result
def test_insufficient_data(self):
"""Very few data points -> returns None."""
prices = _make_daily_prices(2, start_date="2026-02-18", base_price=100.0)
result = analyze_stock(
symbol="TEST",
daily_prices=prices,
earnings_date="2026-02-18",
earnings_timing="bmo",
gap_pct=5.0,
current_price=101.0,
)
assert result is None
# ===========================================================================
# TestModeACandidates (Issue #328: /stable profile returns marketCap, not mktCap)
# ===========================================================================
class TestProfileMarketCap:
def test_prefers_stable_key(self):
assert profile_market_cap({"marketCap": 10, "mktCap": 20}) == 10.0
def test_falls_back_to_legacy_key(self):
assert profile_market_cap({"mktCap": 5e9}) == 5e9
def test_none_stable_uses_legacy(self):
assert profile_market_cap({"marketCap": None, "mktCap": 5e9}) == 5e9
def test_missing_or_none_is_zero(self):
assert profile_market_cap({}) == 0.0
assert profile_market_cap({"marketCap": None}) == 0.0
def test_numeric_string_is_parsed(self):
assert profile_market_cap({"marketCap": "74687168040"}) == 74687168040.0
def test_garbage_is_zero_not_crash(self):
assert profile_market_cap({"marketCap": "n/a"}) == 0.0
assert profile_market_cap({"marketCap": "n/a", "mktCap": 5e9}) == 5e9
def test_negative_is_preserved(self):
assert profile_market_cap({"marketCap": -100}) == -100.0
def test_non_finite_is_zero(self):
"""float("nan")/float("inf") parse cleanly but would bypass the `<` floor."""
assert profile_market_cap({"marketCap": "nan"}) == 0.0
assert profile_market_cap({"marketCap": "inf"}) == 0.0
assert profile_market_cap({"marketCap": float("nan")}) == 0.0
assert profile_market_cap({"marketCap": float("inf"), "mktCap": 5e9}) == 5e9
class TestModeACandidates:
"""_get_candidates_mode_a must keep candidates whose /stable profile has marketCap."""
@staticmethod
def _args(min_market_cap=1_000_000_000, lookback_days=5):
args = MagicMock()
args.min_market_cap = min_market_cap
args.lookback_days = lookback_days
return args
@staticmethod
def _client(profile):
client = MagicMock()
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {"AAPL": profile}
client.get_api_stats.return_value = {"budget_remaining": 100, "rate_limit_reached": False}
return client
def test_issue_328_regression_stable_profile_is_kept(self):
client = self._client(
{"symbol": "AAPL", "marketCap": 3_500_000_000_000, "exchange": "NASDAQ"}
)
result, reason = _get_candidates_mode_a(client, self._args())
assert reason is None
assert len(result) == 1
assert result[0]["symbol"] == "AAPL"
assert result[0]["market_cap"] == 3_500_000_000_000
assert result[0]["earnings_timing"] == "amc"
assert result[0]["gap_pct"] is None
def test_legacy_v3_profile_is_kept(self):
client = self._client(
{"symbol": "AAPL", "mktCap": 3_500_000_000_000, "exchangeShortName": "NASDAQ"}
)
result, reason = _get_candidates_mode_a(client, self._args())
assert len(result) == 1
assert reason is None
def test_below_min_market_cap_is_dropped(self):
client = self._client({"symbol": "AAPL", "marketCap": 999_999_999})
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "all_below_market_cap_floor"
def test_null_market_cap_is_dropped_not_crash(self):
client = self._client({"symbol": "AAPL", "marketCap": None})
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "profiles_missing_required_field:marketCap"
def test_non_numeric_market_cap_is_dropped_not_crash(self):
client = self._client({"symbol": "AAPL", "marketCap": "n/a"})
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "profiles_missing_required_field:marketCap"
def test_missing_profile_is_dropped(self):
client = self._client({"symbol": "AAPL", "marketCap": 3e12})
client.get_company_profiles.return_value = {}
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "no_profiles_returned"
@staticmethod
def _client_with_time(time_value, profile=None):
client = MagicMock()
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": time_value}
]
client.get_company_profiles.return_value = {
"AAPL": profile or {"symbol": "AAPL", "marketCap": 3_500_000_000_000}
}
client.get_api_stats.return_value = {"budget_remaining": 100, "rate_limit_reached": False}
return client
def test_null_time_normalizes_to_unknown_not_empty_string(self):
"""Issue #352: `time: null` (unconfirmed session) must become the
canonical 'unknown', matching what Mode B's validate_input_json
requires -- not an empty string."""
client = self._client_with_time(None)
result, reason = _get_candidates_mode_a(client, self._args())
assert reason is None
assert result[0]["earnings_timing"] == "unknown"
def test_bmo_time_normalizes_to_bmo(self):
client = self._client_with_time("bmo")
result, reason = _get_candidates_mode_a(client, self._args())
assert result[0]["earnings_timing"] == "bmo"
def test_amc_time_normalizes_to_amc(self):
client = self._client_with_time("amc")
result, reason = _get_candidates_mode_a(client, self._args())
assert result[0]["earnings_timing"] == "amc"
def test_missing_time_key_normalizes_to_unknown(self):
client = MagicMock()
client.get_earnings_calendar.return_value = [{"symbol": "AAPL", "date": "2026-09-03"}]
client.get_company_profiles.return_value = {
"AAPL": {"symbol": "AAPL", "marketCap": 3_500_000_000_000}
}
client.get_api_stats.return_value = {"budget_remaining": 100, "rate_limit_reached": False}
result, reason = _get_candidates_mode_a(client, self._args())
assert result[0]["earnings_timing"] == "unknown"
# ===========================================================================
# TestZeroResultReasons (Issue #332: ZERO_RESULT_REASON codes for both modes)
# ===========================================================================
class TestModeAZeroResultReasons:
"""_get_candidates_mode_a reason codes, evaluated in the documented order."""
@staticmethod
def _args(min_market_cap=1_000_000_000, lookback_days=5):
args = MagicMock()
args.min_market_cap = min_market_cap
args.lookback_days = lookback_days
return args
def test_no_earnings_rows_when_calendar_empty(self):
client = MagicMock()
client.get_earnings_calendar.return_value = []
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "no_earnings_rows"
def test_no_earnings_rows_when_no_symbols(self):
client = MagicMock()
client.get_earnings_calendar.return_value = [{"date": "2026-09-03", "time": "amc"}]
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "no_earnings_rows"
def test_calendar_fetch_failed_when_calendar_is_none(self):
"""A failed fetch (None body) must not look like a quiet day."""
client = MagicMock()
client.get_earnings_calendar.return_value = None
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "calendar_fetch_failed"
exit_code, level, _ = _ZERO_RESULT_REASONS[reason]
assert (exit_code, level) == (1, "ERROR")
def test_calendar_fetch_failed_when_calendar_is_non_list(self):
client = MagicMock()
client.get_earnings_calendar.return_value = {"error": "Bad Request"}
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "calendar_fetch_failed"
def test_malformed_rows_are_ignored_not_crash(self):
"""Non-dict rows are never dereferenced; symbols-empty stays benign."""
client = MagicMock()
client.get_earnings_calendar.return_value = ["x", None, {"foo": 1}]
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "no_earnings_rows"
def test_profiles_budget_exhausted_when_budget_remaining_zero(self):
client = MagicMock()
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {}
client.get_api_stats.return_value = {"budget_remaining": 0, "rate_limit_reached": False}
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "profiles_budget_exhausted"
def test_profiles_budget_exhausted_when_rate_limit_reached(self):
client = MagicMock()
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {}
client.get_api_stats.return_value = {"budget_remaining": 5, "rate_limit_reached": True}
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "profiles_budget_exhausted"
def test_no_profiles_returned_when_budget_remains(self):
client = MagicMock()
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {}
client.get_api_stats.return_value = {"budget_remaining": 50, "rate_limit_reached": False}
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "no_profiles_returned"
def test_missing_required_field_market_cap_when_key_absent_from_all_profiles(self):
client = MagicMock()
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {"AAPL": {"exchange": "NASDAQ"}}
client.get_api_stats.return_value = {"budget_remaining": 50, "rate_limit_reached": False}
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "profiles_missing_required_field:marketCap"
def test_all_below_market_cap_floor_when_key_present_but_low(self):
client = MagicMock()
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {
"AAPL": {"marketCap": 1_000, "exchange": "NASDAQ"}
}
client.get_api_stats.return_value = {"budget_remaining": 50, "rate_limit_reached": False}
result, reason = _get_candidates_mode_a(client, self._args(min_market_cap=1_000_000_000))
assert result == []
assert reason == "all_below_market_cap_floor"
def test_missing_required_field_market_cap_when_all_values_null(self):
"""Key present but null/non-numeric for every profile must still be
classified as missing-field, not as an ordinary below-floor day
(review round 1)."""
client = MagicMock()
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"},
{"symbol": "MSFT", "date": "2026-09-03", "time": "amc"},
]
client.get_company_profiles.return_value = {
"AAPL": {"marketCap": None, "exchange": "NASDAQ"},
"MSFT": {"marketCap": "n/a", "exchange": "NASDAQ"},
}
client.get_api_stats.return_value = {"budget_remaining": 50, "rate_limit_reached": False}
result, reason = _get_candidates_mode_a(client, self._args())
assert result == []
assert reason == "profiles_missing_required_field:marketCap"
class TestModeBZeroResultReasons:
def test_no_input_candidates_when_grade_filter_excludes_all(self, tmp_path):
payload = {
"schema_version": "1.0",
"results": [
{
"symbol": "AAPL",
"earnings_date": "2026-09-03",
"earnings_timing": "amc",
"gap_pct": 5.0,
"grade": "D",
}
],
}
json_path = tmp_path / "candidates.json"
json_path.write_text(json.dumps(payload))
args = MagicMock()
args.candidates_json = str(json_path)
args.min_grade = "A"
result, reason = _get_candidates_mode_b(args)
assert result == []
assert reason == "no_input_candidates"
def test_non_empty_candidates_have_no_reason(self, tmp_path):
payload = {
"schema_version": "1.0",
"results": [
{
"symbol": "AAPL",
"earnings_date": "2026-09-03",
"earnings_timing": "amc",
"gap_pct": 5.0,
"grade": "A",
}
],
}
json_path = tmp_path / "candidates.json"
json_path.write_text(json.dumps(payload))
args = MagicMock()
args.candidates_json = str(json_path)
args.min_grade = "B"
result, reason = _get_candidates_mode_b(args)
assert len(result) == 1
assert reason is None
class TestFixtureConsumerFieldContracts:
"""D4: pin profile_market_cap and the Mode A marketCap reason against a
sanitized live /stable/profile fixture row (Issue #332)."""
FIXTURE_PROFILE = {
"symbol": "AAPL",
"price": 319.97,
"marketCap": 4699513299320,
"beta": 1.086,
"lastDividend": 1.06,
"exchangeFullName": "NASDAQ Global Select",
"exchange": "NASDAQ",
"industry": "Consumer Electronics",
"sector": "Technology",
"country": "US",
}
def test_profile_market_cap_matches_fixture(self):
assert profile_market_cap(self.FIXTURE_PROFILE) == 4699513299320
def test_profile_market_cap_legacy_alias(self):
assert profile_market_cap({"mktCap": 5e9}) == 5e9
def test_profile_market_cap_empty_profile_is_zero(self):
assert profile_market_cap({}) == 0.0
def test_mode_a_reason_when_marketcap_renamed(self):
renamed = dict(self.FIXTURE_PROFILE)
renamed["mktCap"] = renamed.pop("marketCap")
# mktCap alias keeps the candidate (legacy key path), so drop the
# legacy key entirely to reproduce the #328 field-rename signature.
del renamed["mktCap"]
client = MagicMock()
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {"AAPL": renamed}
client.get_api_stats.return_value = {"budget_remaining": 50, "rate_limit_reached": False}
args = MagicMock()
args.min_market_cap = 2_000_000_000
args.lookback_days = 5
result, reason = _get_candidates_mode_a(client, args)
assert result == []
assert reason == "profiles_missing_required_field:marketCap"
class TestTimingMetadata:
"""Issue #352: metadata carries timing_unknown_count / timing_candidates_total
/ timing_source for Mode A, and None for all three in Mode B."""
@patch("screen_pead.FMPClient")
def test_mode_a_reports_timing_unknown_count_and_source(self, mock_client_class, tmp_path):
client = mock_client_class.return_value
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "bmo"},
{"symbol": "MSFT", "date": "2026-09-03", "time": "amc"},
{"symbol": "GOOG", "date": "2026-09-03", "time": None},
]
client.get_company_profiles.return_value = {
"AAPL": {"marketCap": 3e12, "exchange": "NASDAQ"},
"MSFT": {"marketCap": 2e12, "exchange": "NASDAQ"},
"GOOG": {"marketCap": 1.5e12, "exchange": "NASDAQ"},
}
prices = [
{
"date": "2026-09-03",
"open": 100.0,
"high": 101.0,
"low": 99.0,
"close": 100.5,
"volume": 1_000_000,
}
] * 10
client.get_historical_prices.return_value = {"historical": prices}
client.get_api_stats.return_value = {
"api_calls_made": 4,
"budget_remaining": 100,
"rate_limit_reached": False,
}
argv = ["screen_pead.py", "--api-key", "test-key", "--output-dir", str(tmp_path)]
with patch.object(sys, "argv", argv):
main()
json_files = list(tmp_path.glob("pead_screener_*.json"))
assert len(json_files) == 1
data = json.loads(json_files[0].read_text())
metadata = data["metadata"]
assert metadata["timing_candidates_total"] == 3
assert metadata["timing_unknown_count"] == 1
assert metadata["timing_source"] == "fmp_stable_includeReportTimes"
@patch("screen_pead._get_candidates_mode_a")
@patch("screen_pead.FMPClient")
def test_mode_a_timing_counts_only_candidates_surviving_budget_trim(
self, mock_client_class, mock_get_candidates, tmp_path
):
"""Regression: timing_candidates_total/timing_unknown_count must be
counted AFTER the Phase 1.5 budget trim, not before it -- otherwise
the reported denominator includes candidates that never reach
Phase 2 analysis at all."""
mock_get_candidates.return_value = (
[
{
"symbol": "A",
"earnings_date": "2026-09-03",
"earnings_timing": "bmo",
"gap_pct": None,
"market_cap": 5e9,
},
{
"symbol": "B",
"earnings_date": "2026-09-03",
"earnings_timing": "unknown",
"gap_pct": None,
"market_cap": 4e9,
},
{
"symbol": "C",
"earnings_date": "2026-09-03",
"earnings_timing": "unknown",
"gap_pct": None,
"market_cap": 3e9,
},
{
"symbol": "D",
"earnings_date": "2026-09-03",
"earnings_timing": "amc",
"gap_pct": None,
"market_cap": 2e9,
},
],
None,
)
client = mock_client_class.return_value
prices = [
{
"date": "2026-09-03",
"open": 100.0,
"high": 101.0,
"low": 99.0,
"close": 100.5,
"volume": 1_000_000,
}
] * 10
client.get_historical_prices.return_value = {"historical": prices}
# budget_remaining (2) is lower than the candidate count (4), so the
# Phase 1.5 trim branch executes and keeps only the first 2 (A, B).
client.get_api_stats.return_value = {
"api_calls_made": 1,
"budget_remaining": 2,
"rate_limit_reached": False,
}
argv = ["screen_pead.py", "--api-key", "test-key", "--output-dir", str(tmp_path)]
with patch.object(sys, "argv", argv):
main()
json_files = list(tmp_path.glob("pead_screener_*.json"))
assert len(json_files) == 1
data = json.loads(json_files[0].read_text())
metadata = data["metadata"]
# Trimmed population is [A (bmo), B (unknown)], not the original 4.
assert metadata["timing_candidates_total"] == 2
assert metadata["timing_unknown_count"] == 1
def test_mode_b_leaves_timing_metadata_none(self, tmp_path):
payload = {
"schema_version": "1.0",
"results": [
{
"symbol": "AAPL",
"earnings_date": "2026-09-03",
"earnings_timing": "amc",
"gap_pct": 5.0,
"grade": "A",
}
],
}
json_path = tmp_path / "candidates.json"
json_path.write_text(json.dumps(payload))
with patch("screen_pead.FMPClient") as mock_client_class:
client = mock_client_class.return_value
prices = [
{
"date": "2026-09-03",
"open": 100.0,
"high": 101.0,
"low": 99.0,
"close": 100.5,
"volume": 1_000_000,
}
] * 10
client.get_historical_prices.return_value = {"historical": prices}
client.get_api_stats.return_value = {
"api_calls_made": 1,
"budget_remaining": 100,
"rate_limit_reached": False,
}
argv = [
"screen_pead.py",
"--api-key",
"test-key",
"--output-dir",
str(tmp_path),
"--candidates-json",
str(json_path),
]
with patch.object(sys, "argv", argv):
main()
json_files = list(tmp_path.glob("pead_screener_*.json"))
assert len(json_files) == 1
data = json.loads(json_files[0].read_text())
metadata = data["metadata"]
assert metadata["timing_candidates_total"] is None
assert metadata["timing_unknown_count"] is None
assert metadata["timing_source"] is None
class TestMainZeroResultExitCodes:
"""Drive main() end-to-end with a mocked FMPClient for each exit code."""
@staticmethod
def _argv(tmpdir, extra=None):
argv = ["screen_pead.py", "--api-key", "test-key", "--output-dir", str(tmpdir)]
return argv + (extra or [])
@patch("screen_pead.FMPClient")
def test_mode_a_no_earnings_rows_exits_0(self, mock_client_class, tmp_path, capsys):
client = mock_client_class.return_value
client.get_earnings_calendar.return_value = []
with patch.object(sys, "argv", self._argv(tmp_path)):
with pytest.raises(SystemExit) as exc_info:
main()
assert exc_info.value.code == 0
err = capsys.readouterr().err
assert "ZERO_RESULT_REASON=no_earnings_rows" in err
@patch("screen_pead.FMPClient")
def test_mode_a_no_profiles_returned_exits_1(self, mock_client_class, tmp_path, capsys):
client = mock_client_class.return_value
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {}
client.get_api_stats.return_value = {"budget_remaining": 50, "rate_limit_reached": False}
with patch.object(sys, "argv", self._argv(tmp_path)):
with pytest.raises(SystemExit) as exc_info:
main()
assert exc_info.value.code == 1
err = capsys.readouterr().err
assert "ZERO_RESULT_REASON=no_profiles_returned" in err
@patch("screen_pead.FMPClient")
def test_mode_a_profiles_budget_exhausted_exits_0(self, mock_client_class, tmp_path, capsys):
client = mock_client_class.return_value
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {}
client.get_api_stats.return_value = {"budget_remaining": 0, "rate_limit_reached": False}
with patch.object(sys, "argv", self._argv(tmp_path)):
with pytest.raises(SystemExit) as exc_info:
main()
assert exc_info.value.code == 0
err = capsys.readouterr().err
assert "ZERO_RESULT_REASON=profiles_budget_exhausted" in err # pragma: allowlist secret
@patch("screen_pead.FMPClient")
def test_mode_a_missing_marketcap_field_exits_1(self, mock_client_class, tmp_path, capsys):
client = mock_client_class.return_value
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {"AAPL": {"exchange": "NASDAQ"}}
client.get_api_stats.return_value = {"budget_remaining": 50, "rate_limit_reached": False}
with patch.object(sys, "argv", self._argv(tmp_path)):
with pytest.raises(SystemExit) as exc_info:
main()
assert exc_info.value.code == 1
err = capsys.readouterr().err
assert "ZERO_RESULT_REASON=profiles_missing_required_field:marketCap" in err
@patch("screen_pead.FMPClient")
def test_mode_a_all_below_market_cap_floor_exits_0(self, mock_client_class, tmp_path, capsys):
client = mock_client_class.return_value
client.get_earnings_calendar.return_value = [
{"symbol": "AAPL", "date": "2026-09-03", "time": "amc"}
]
client.get_company_profiles.return_value = {
"AAPL": {"marketCap": 1_000, "exchange": "NASDAQ"}
}
client.get_api_stats.return_value = {"budget_remaining": 50, "rate_limit_reached": False}
with patch.object(sys, "argv", self._argv(tmp_path, ["--min-market-cap", "1000000000"])):
with pytest.raises(SystemExit) as exc_info:
main()
assert exc_info.value.code == 0
err = capsys.readouterr().err
assert "ZERO_RESULT_REASON=all_below_market_cap_floor" in err
@patch("screen_pead.FMPClient")
def test_mode_b_no_input_candidates_exits_0(self, mock_client_class, tmp_path, capsys):
payload = {
"schema_version": "1.0",
"results": [
{
"symbol": "AAPL",
"earnings_date": "2026-09-03",
"earnings_timing": "amc",
"gap_pct": 5.0,
"grade": "D",
}
],
}
json_path = tmp_path / "candidates.json"
json_path.write_text(json.dumps(payload))
with patch.object(
sys,
"argv",
self._argv(tmp_path, ["--candidates-json", str(json_path), "--min-grade", "A"]),
):
with pytest.raises(SystemExit) as exc_info:
main()
assert exc_info.value.code == 0
err = capsys.readouterr().err
assert "ZERO_RESULT_REASON=no_input_candidates" in err