references/distribution_day_guide.md
# Distribution Day Guide
## What is a Distribution Day?
A Distribution Day occurs when a major stock index (S&P 500 or NASDAQ Composite) declines by 0.2% or more on volume that is higher than the previous trading day. This pattern indicates institutional investors are selling stocks in volume - "distributing" their holdings.
William O'Neil introduced this concept in "How to Make Money in Stocks" as one of the most reliable signals that a market uptrend is under pressure.
## Precise Rules
### Distribution Day Criteria (BOTH must be met):
1. **Price decline >= 0.2%** from previous session's close
2. **Volume > previous day's volume** (even slightly higher counts)
### Stalling Day Criteria (BOTH must be met):
1. **Volume > previous day's volume** (institutional activity present)
2. **Price gain < 0.1%** (price fails to advance meaningfully despite volume)
3. Stalling days count as **0.5 distribution days** in the effective count
### 25-Day Expiration Rule:
- A distribution day automatically expires after **25 trading days** (approximately 5 weeks)
- Only the rolling 25-day window matters
- This prevents old distribution from affecting current assessment
### What Does NOT Count as Distribution:
- Index declines on lower volume (normal profit-taking)
- Large gap-up open followed by slight closing decline (still net positive day)
- Volume below average (lack of institutional participation)
## Counting and Interpretation
| Effective Count | Interpretation | Action |
|----------------|----------------|--------|
| 0-1 | Healthy market | Normal operations |
| 2-3 | Some selling but normal | Monitor closely |
| 4 | O'Neil's initial warning | Tighten stops |
| 5 | Significant distribution | Reduce exposure |
| 6+ | Heavy distribution | Protect capital aggressively |
**Effective Count Formula:**
```
Effective Count = Distribution Days + (0.5 × Stalling Days)
```
## Dual-Index Approach
This tool monitors both the S&P 500 and NASDAQ (via QQQ). The higher effective count between the two indices is used for scoring, because:
1. Distribution can concentrate in one index before the other
2. NASDAQ often leads due to higher growth stock concentration
3. The worst signal from either index is the most conservative approach
## Common Patterns at Market Tops
### Classic Distribution Sequence (2022 Example):
1. Week 1: One distribution day (seems harmless)
2. Week 2: Two more distribution days
3. Week 3: One stalling day (selling into rallies)
4. Week 4: Two more distribution days (total: 5-6 effective)
5. Market breaks down shortly after
### False Alarm Pattern:
1. Two distribution days in quick succession
2. Market rallies strongly, erasing concerns
3. Distribution days expire without reaching critical mass
4. Uptrend continues
### Key Distinction:
The **clustering** of distribution days matters. Three distribution days in one week is more alarming than three spread over five weeks.
## Follow-Through Day (FTD)
The FTD is the counterpart to distribution days - it signals a potential new uptrend after a correction.
### FTD Rules:
1. **Rally Attempt:** Market makes its low and begins to rally (first up day)
2. **Counting:** Start counting from day 1 of the rally attempt
3. **FTD Occurs on Day 4-7:** A significant price gain (>1.5%) with volume higher than the previous day
4. **Confirmation:** The FTD should be on day 4 at the earliest (premature rallies on days 1-3 are unreliable)
### FTD Reliability:
- Not all FTDs lead to sustained uptrends (~25% fail)
- Multiple FTDs within a short period increase confidence
- FTDs work best when combined with improving breadth
- O'Neil: "I never try to buy at the bottom. I wait for the Follow-Through Day."
## Practical Tips
### When Monitoring Distribution:
1. Check distribution count DAILY during uncertain markets
2. Pay special attention when count reaches 3 (pre-warning)
3. At 4+, begin defensive actions regardless of other signals
4. Don't wait for 6+ to act - by then, significant damage may be done
### Common Mistakes:
1. **Ignoring stalling days:** They're subtle but significant
2. **Resetting count manually:** Trust the 25-day expiration rule
3. **Using wrong volume data:** Must compare day-to-day, not vs average
4. **Mixing indices:** Count separately for S&P 500 and NASDAQ
5. **Ignoring context:** 4 distribution days in a strong uptrend is different from 4 in a range-bound market
references/historical_tops.md
# Historical Market Top Analysis
## Overview
Analysis of major market tops through the lens of the 6-component Market Top Detector framework. These historical cases are retrospective illustrations with estimated component scores, not point-in-time calibration data.
## 2000: Dot-Com Bubble Top (March 2000)
### Timeline
- **S&P 500 Peak:** March 24, 2000 (1,527.46)
- **NASDAQ Peak:** March 10, 2000 (5,048.62)
- **Subsequent Decline:** S&P -49%, NASDAQ -78% over 2.5 years
### Component Signals (Estimated Scores)
| Component | Signal Level | Est. Score | Notes |
|-----------|-------------|------------|-------|
| Distribution Days | Very High | 90-100 | Heavy distribution throughout March 2000 |
| Leading Stocks | Critical | 85-95 | Internet stocks peaked weeks before indices |
| Defensive Rotation | Strong | 70-80 | Value/defensive outperformed for months |
| Breadth | Critical Divergence | 80-90 | Advance/Decline line peaked in April 1998 |
| Index Technical | Breaking Down | 70-80 | NASDAQ broke 50DMA first |
| Sentiment | Extreme | 90-100 | IPO mania, "new paradigm" narrative |
**Estimated Composite: 82-92 (Critical)**
### Key Lesson
Breadth divergence was the earliest signal (nearly 2 YEARS before the top). Distribution days and leading stock breakdown were the final confirmation signals. By the time all 6 components were flashing red, the decline had already begun.
---
## 2007: Financial Crisis Top (October 2007)
### Timeline
- **S&P 500 Peak:** October 9, 2007 (1,565.15)
- **Subsequent Decline:** -57% over 17 months (March 2009 low)
### Component Signals (Estimated Scores)
| Component | Signal Level | Est. Score | Notes |
|-----------|-------------|------------|-------|
| Distribution Days | High | 75-85 | Multiple distribution clusters in Sept-Oct 2007 |
| Leading Stocks | High | 70-80 | Homebuilders (ITB) peaked in Feb 2006, financials (XLF) peaked in June 2007 |
| Defensive Rotation | Moderate | 55-65 | Utilities and healthcare outperforming |
| Breadth | Diverging | 60-70 | NYSE advance/decline line peaked July 2007 |
| Index Technical | Weakening | 50-60 | S&P still near highs but internal structure deteriorating |
| Sentiment | Elevated | 55-65 | VIX was very low (sub-12) for extended period in 2006-2007 |
**Estimated Composite: 62-72 (Red Zone)**
### Key Lesson
The 2007 top was a slow-motion process. Leading sector (financials/housing) peaked 12-18 months before the S&P 500. The composite would have been in the "Orange" zone for months before reaching "Red." This top rewarded patience in reducing exposure rather than a single binary call.
---
## 2018: Q4 Sell-Off (September-December 2018)
### Timeline
- **S&P 500 Peak:** September 20, 2018 (2,930.75)
- **Subsequent Decline:** -20% (correction, not bear market)
- **Recovery:** V-bottom in December 2018, new highs by April 2019
### Component Signals (Estimated Scores)
| Component | Signal Level | Est. Score | Notes |
|-----------|-------------|------------|-------|
| Distribution Days | Moderate | 55-65 | Distribution accumulated through October |
| Leading Stocks | Moderate | 50-60 | FAANG stocks peaked at different times (FB in July, AMZN in September) |
| Defensive Rotation | Moderate | 45-55 | Utilities outperformed in Q3 |
| Breadth | Mild Divergence | 40-50 | Breadth narrowing but not extreme |
| Index Technical | Weakening | 55-65 | 50DMA crossed below 200DMA in December |
| Sentiment | Mixed | 35-45 | VIX was elevated but not extreme |
**Estimated Composite: 47-57 (Orange Zone)**
### Key Lesson
This was a classic "correction" case (not a bear market). The composite correctly would have been in Orange zone - enough to warrant profit-taking but not full defensive posture. The rapid V-recovery demonstrated why the composite avoids going to "Critical" without extreme readings across multiple components.
---
## 2022: Fed Tightening Bear Market (January 2022)
### Timeline
- **S&P 500 Peak:** January 3, 2022 (4,796.56)
- **NASDAQ Peak:** November 19, 2021 (16,057.44)
- **Subsequent Decline:** S&P -25%, NASDAQ -33%
### Component Signals (Estimated Scores at January 2022)
| Component | Signal Level | Est. Score | Notes |
|-----------|-------------|------------|-------|
| Distribution Days | High | 75-90 | Heavy distribution in late December 2021 and early January 2022 |
| Leading Stocks | Critical | 80-90 | ARKK peaked Feb 2021 (11 months early), many growth stocks -50%+ from highs |
| Defensive Rotation | Strong | 65-75 | XLU, XLP outperforming strongly |
| Breadth | Severe Divergence | 70-80 | NASDAQ new highs/new lows diverging sharply |
| Index Technical | Breaking | 60-70 | NASDAQ broke 200DMA, S&P testing key levels |
| Sentiment | Elevated | 50-60 | Margin debt at record highs, crypto speculation |
**Estimated Composite: 68-78 (Red Zone)**
### Key Lesson
ARKK (leading stock proxy) peaked 11 months before S&P 500. This is a textbook Minervini signal - leading stocks deteriorate long before the index. By January 2022, the composite would have been solidly in the Red zone, correctly signaling aggressive profit-taking. The 2022 case validates the "Leading Stock Health" component as an early warning signal.
---
## Cross-Case Patterns
### What All Tops Had in Common:
1. **Distribution days clustered** (not spread out) in the weeks before the decline
2. **Leading stocks peaked before the index** (by weeks to months)
3. **Breadth diverged** from index price (the rally became increasingly narrow)
4. **Defensive rotation** accelerated as the top formed
### What Varied:
1. **Severity:** Ranged from -20% correction (2018) to -78% NASDAQ crash (2000)
2. **Speed:** 2018 was a fast decline (3 months), 2000-2002 was slow (2.5 years)
3. **Recovery time:** 2018 recovered in 4 months, 2000 took 7 years (NASDAQ)
4. **Breadth divergence timing:** 2000 had 2-year divergence, 2022 had months
### Illustrative Score Patterns:
- **Orange Zone (41-60):** Retrospective correction illustration (2018-type, -10-20%)
- **Red Zone (61-80):** Retrospective major-decline illustration (2007, 2022-type, -20-35%)
- **Critical (81-100):** Retrospective bubble-top illustration (2000-type, -40%+)
### Important Caveat:
These retrospective examples do not establish measured calibration. Each market environment is unique. The composite score should be used as one input into investment decisions, not as a definitive prediction. Markets can remain in "warning" zones for extended periods before actually declining.
---
## Pattern: Leading Stocks as Early Warning
From the Monty article:
> 弱気相場の初期段階では、特定の主導株が下降トレンドに抵抗するかのように強く、
> 上昇できるという印象を与えます。
This pattern appeared in every historical top:
- **2000:** Blue-chip tech (MSFT, INTC) held up while dot-coms crashed
- **2007:** Energy sector rallied while financials imploded
- **2022:** AAPL, MSFT relatively stable while ARKK stocks crashed
- **Current analog:** Some mega-caps hold while growth/speculative names deteriorate
This is why Component 2 (Leading Stock Health) uses a broad basket of growth ETFs rather than just the S&P 500 components - it captures the deterioration that indices mask.
references/market_top_methodology.md
# Market Top Detection Methodology
## Overview
This framework integrates three approaches to market top detection, each targeting different aspects of institutional behavior that precede significant market declines (10-20% corrections).
## The Three Pillars
### Pillar 1: O'Neil - Distribution Day Accumulation
**Source:** William O'Neil, "How to Make Money in Stocks"
**Core Concept:** Institutional investors (mutual funds, hedge funds, pensions) cannot sell their large positions in a single day. Their selling creates a distinctive pattern: the index declines on higher-than-previous-day volume. When these "distribution days" accumulate, it signals that smart money is systematically exiting.
**Distribution Day Definition:**
- Index declines >= 0.2% from previous close
- Volume is higher than previous day's volume
- Both conditions must be met simultaneously
**Stalling Day Definition:**
- Volume increases from previous day
- But price gain is minimal (< 0.1%)
- Indicates institutional selling into strength
- Counted at half weight (0.5 distribution days)
**25-Day Window:**
- Distribution days expire after 25 trading days
- Only the rolling 25-day count matters
- This prevents old events from distorting current assessment
**O'Neil's Warning Thresholds:**
- 4-5 distribution days in 25 trading days = "market under pressure"
- 6+ distribution days = "distribution is heavy, protect capital"
- Combined with stalling days for full picture
### Pillar 2: Minervini - Leading Stock Deterioration
**Source:** Mark Minervini, "Trade Like a Stock Market Wizard" / "Think & Trade Like a Champion"
**Core Concept:** Market tops do not appear suddenly. They develop through a process where leading stocks - the strongest performers of the previous rally - begin breaking down before the major indices. This is because institutional investors sell their biggest winners first to lock in profits.
**Key Observation from Monty Article:**
> "弱気相場の初期段階では、特定の主導株が下降トレンドに抵抗するかのように強く、上昇できるという印象を与えます。"
> (In the early stages of a bear market, certain leading stocks appear to resist the downtrend, giving the impression they can still rise.)
**Detection Method:**
Using a basket of growth/innovation ETFs as proxy for market leadership:
- ARKK (Innovation), WCLD (Cloud), IGV (Software)
- XBI (Biotech), SOXX/SMH (Semiconductors)
- KWEB (China Tech), TAN (Solar)
**Deterioration Signals per ETF:**
1. Distance from 52-week high (>10% = concerning, >25% = bear territory)
2. Price below 50-day moving average
3. Price below 200-day moving average
4. Formation of lower highs pattern
**Amplification Rule:** When 60%+ of leading ETFs show deterioration, the signal is amplified by 1.3x. This reflects the systemic nature of leadership breakdown.
### Pillar 3: Monty - Defensive Sector Rotation
**Source:** monty-trader.com "米国株 株式相場の天井の見極め方と下落局面でやるべきこと"
**Core Concept:** Before a market top, capital flows from offensive/growth sectors into defensive/value sectors. This "rotation" occurs because institutional investors are becoming defensive while maintaining their equity allocation. It's a critical early warning signal.
**Defensive Sectors (Safety Seekers):**
- XLU (Utilities) - stable cash flows, bond-like
- XLP (Consumer Staples) - recession-resistant demand
- XLV (Healthcare) - non-discretionary spending
- VNQ (Real Estate) - income-focused
**Offensive Sectors (Growth/Risk):**
- XLK (Technology) - growth-dependent
- XLC (Communication Services) - ad-spend sensitive
- XLY (Consumer Discretionary) - economically sensitive
- QQQ (NASDAQ 100) - tech-heavy growth proxy
**Signal:** When defensive sectors outperform offensive sectors over a 20-day rolling period, capital is flowing defensively. A relative performance spread of +3% or more is a strong warning signal.
---
## Supplementary Components
### Component 4: Market Breadth Divergence
**Concept:** When the index makes new highs but fewer stocks participate, the rally is narrowing and vulnerable. This "divergence" between index price and market breadth is a classic top signal.
**Key Metric:** Percentage of S&P 500 stocks above their 200-day moving average.
- Healthy market: >70% of stocks above 200DMA
- Warning zone: 50-70% with index near highs
- Critical: <50% with index at highs
**Important Nuance:** Breadth divergence is most meaningful when the index is near its 52-week high (within -5%). If the index has already corrected significantly, weak breadth is expected and less informative.
### Component 5: Index Technical Condition
**Moving Average Structure:**
- Healthy: Price > 21 EMA > 50 EMA > 200 SMA
- Deteriorating: Any of these relationships inverting
- Bearish: Price below all major moving averages
**Pattern Recognition:**
- Failed Rally: Price bounces but fails to exceed recent peak
- Lower Highs: Sequential swing highs declining
- Gap Downs on Volume: Institutional panic selling
### Component 6: Sentiment & Speculation
**VIX (Fear Index):**
- <12: Extreme complacency (warning)
- 12-16: Low volatility (mild warning)
- 16-20: Normal
- >25: Fear elevated (top less likely, correction may be underway)
**Put/Call Ratio:**
- <0.60: Extreme call buying (maximum complacency)
- 0.60-0.70: Elevated optimism
- 0.70-0.80: Mildly bullish
- >0.80: Healthy caution
**VIX Term Structure:**
- Steep Contango: Market expects calm (complacency signal)
- Backwardation: Hedging demand elevated (fear present)
---
## Follow-Through Day (FTD) Monitor
O'Neil's concept for bottom confirmation, relevant when composite score > 40:
1. **Rally Attempt Day:** First day the index closes up after a decline
2. **Counting begins:** Day 1 of potential new uptrend
3. **Follow-Through Day:** On days 4-7 of rally attempt, a strong gain (1.5%+) on higher volume than previous day
4. **Significance:** "The most powerful uptrends usually begin with a Follow-Through Day on day 4-7"
**False FTD Rate:** Approximately 25% of FTDs fail. Multiple FTDs increase confidence.
---
## Scoring Philosophy
### Why Weighted Composite?
No single indicator perfectly predicts market tops. Each pillar captures a different dimension:
- Distribution Days: Direct measurement of institutional selling
- Leading Stocks: Quality of market leadership
- Defensive Rotation: Institutional positioning shift
- Breadth: Market participation health
- Technicals: Price structure integrity
- Sentiment: Psychological extremes
### Weight Rationale
- **Distribution Days (25%):** Most direct measure of institutional behavior
- **Leading Stocks (20%):** Minervini's strongest conviction signal
- **Defensive Rotation (15%):** Monty's key differentiator
- **Breadth (15%):** Classic confirmation signal
- **Technicals (15%):** Structural integrity check
- **Sentiment (10%):** Context and extremes (less reliable alone)
### Heuristic Design Targets
The heuristic scoring was designed around the following illustrative targets; these are not measured calibration results:
- A "normal" healthy market scores 10-20
- An "early warning" market scores 25-40
- A market in the "initial to middle stage of top formation" scores 40-55
- A clear top formation scores 60-80
- A completed top with breakdown scores 80+
references/point_in_time_evaluation.md
# Point-in-time evaluation protocol v1
Status: specification only; no dataset or measured calibration results are supplied.
The existing four historical component vectors are illustrative retrospective
estimates and must never be substituted for archived observations.
## Frozen estimand and score
Before collecting outcomes, freeze a manifest with protocol version `market-top-pit-v1`,
score version `market-top-risk-v1`, code commit, configuration hash, provider versions,
universe-selection rules, missing-data policy and calendar version. Preserve the
current weights, correlation discount and availability-based redistribution in the
score implementation. Record all six raw components, availability flags, adjusted
components, effective weights and final score. Do not pool different availability
masks or score versions when reporting band frequencies.
Define the primary event as an S&P 500 closing-price loss of at least 10% relative
to the as-of close at any of the next 40 trading-session closes. This operational
correction event does not identify a literal local market top. Freeze this endpoint;
other horizons (10/20 sessions) and thresholds (15/20%) are secondary and reported
separately, without replacing a disappointing primary result.
## Versioned dataset specification
Store immutable UTF-8 JSONL observations and a JSON manifest with file SHA-256 hashes.
One observation per Friday session close (last session of a holiday week); sample
all eligible weeks, including ordinary/non-event periods, without top-based selection.
Each row contains:
- `schema_version`: `market-top-pit-v1`; unique `observation_id`, `as_of` (UTC),
exchange session date, score version, code commit and configuration hash.
- `inputs`: immutable artifact URI and SHA-256 for every input; source, vintage,
publication time, retrieval time, original timezone and maximum observation time.
Every observation/publication must be no later than `as_of`. Retrieval after
`as_of` is acceptable only for a verifiable archived vintage available then.
- Point-in-time basket membership and selection evidence (including delisted names);
six raw scores and availability flags, adjustments, effective weights and final score.
- `split`: development or holdout; `label_status`: pending, complete or excluded;
forward 40 session dates/closes, as-of close, minimum forward return and event boolean.
Outcome data lives separately from scoring inputs and is never read by scoring.
- Exclusion reason for unverifiable vintage, missing calendar/prices, unavailable
required inputs or incomplete horizon. Never encode excluded/pending as non-events.
No historical live `--as-of` run is valid evidence: the current CLI rejects it because
live quotes are not archived point-in-time quotes. A future offline replay must read
only hashed archived inputs and reproduce component/score outputs before evaluation.
## Chronological evaluation and leakage controls
Pre-register a date range and split: development 2000-01-01 through 2015-12-31;
holdout 2016-01-01 through 2025-12-31. Purge development observations whose 40-session
outcome crosses the holdout boundary. If these dates cannot be reconstructed from
verifiable vintages, report infeasible and pre-register a new untouched prospective
holdout; do not manufacture history. These periods are protocol choices, not results.
Fit any thresholds solely on development data, freeze hashes before opening holdout
outcomes, and evaluate the holdout once. Record previous exposure to holdout data;
if already used for tuning it cannot be described as untouched.
## Reproducibility and reporting contract
A future evaluator must validate input hashes/timestamps, reconstruct scores from
archived inputs, label complete horizons, enforce split/purge rules, then emit a
versioned JSON metrics artifact plus a human report. Fixed inputs and code must
produce identical rows/metrics; missing vintages stop that observation with a reason.
Use fixed score bands [0,20], (20,40], (40,60], (60,80], (80,100]. For each split,
availability mask and band, report n, events, non-events, empirical event rate and
95% Wilson interval. These are observed frequencies, not a new probability model.
At a predeclared score >60 alert threshold report TP/FP/FN/TN, false alarms (FP),
misses (FN), precision, recall and coverage; undefined denominators yield null.
Report pending/excluded counts and reasons separately. Since weekly 40-session
outcomes overlap, Wilson intervals are descriptive only: also report uncertainty
using a moving-block bootstrap (52-week blocks, 2,000 replicates, fixed seed 515),
flag insufficient blocks/samples, and publish split-level counts. Never imply
independent trials. Publish non-event and event examples with provenance.
A trained probability mapping requires a separate version, development-only fitting
and untouched holdout calibration metrics. Until those measured results exist, retain
`calibrated_probability: false`. The 15-point analogue tolerance is heuristic and
cannot be tuned against these illustrative historical vectors to claim validation.
requirements.txt
# GENERATED by scripts/generate_market_calendar.py — do not edit.
# Source of truth: scripts/market_calendar/ and config/market-calendar-dependency.yaml.
pandas-market-calendars==5.2.2
requests>=2.31.0
yfinance>=0.2.0 # optional: FMP success path never depends on it; fallback returns None
scripts/_market_calendar.py
# GENERATED by scripts/generate_market_calendar.py — do not edit.
# Source of truth: scripts/market_calendar/ and config/market-calendar-dependency.yaml.
"""Shared exchange-calendar and timezone contract for cash-equity skills.
The module is vendored into each standalone skill by
``scripts/generate_market_calendar.py``. Runtime code must not read repository
configuration: packaged skills contain this self-contained module plus an exact
runtime dependency declaration.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from typing import Any, Optional
from zoneinfo import ZoneInfo
class CalendarUnavailableError(RuntimeError):
"""Raised when an exchange calendar cannot be loaded or queried safely."""
@dataclass(frozen=True)
class Session:
"""One regular cash-equity session in the venue's IANA timezone."""
venue: str
session_date: date
timezone: str
market_open: datetime
market_close: datetime
break_start: Optional[datetime] = None
break_end: Optional[datetime] = None
_VENUES = {
"XNYS": ("NYSE", "America/New_York"),
# pandas-market-calendars exposes NASDAQ as an alias of its NYSE calendar.
# Both US cash venues intentionally share this session set.
"XNAS": ("NASDAQ", "America/New_York"),
"XTKS": ("JPX", "Asia/Tokyo"),
"XLON": ("LSE", "Europe/London"),
}
def supported_venues() -> tuple[str, ...]:
"""Return supported venue identifiers in deterministic order."""
return tuple(sorted(_VENUES))
def _venue_contract(venue: str) -> tuple[str, str, str]:
if not isinstance(venue, str):
raise ValueError("venue must be a string")
normalized = venue.strip().upper()
try:
provider_name, timezone_name = _VENUES[normalized]
except KeyError as exc:
raise ValueError(
f"Unsupported venue {venue!r}; expected one of {', '.join(supported_venues())}"
) from exc
return normalized, provider_name, timezone_name
def _require_date(value: date, *, name: str) -> date:
if type(value) is not date:
raise TypeError(f"{name} must be datetime.date (not datetime)")
return value
def require_aware(value: datetime, *, name: str = "timestamp") -> datetime:
"""Reject naive datetimes at the shared boundary."""
if not isinstance(value, datetime) or value.tzinfo is None:
raise ValueError(f"{name} must be a timezone-aware datetime")
if value.utcoffset() is None:
raise ValueError(f"{name} must have a resolvable UTC offset")
return value
def parse_aware_datetime(value: str) -> datetime:
"""Parse an offset-bearing ISO-8601 timestamp, accepting a trailing ``Z``."""
if not isinstance(value, str) or not value.strip():
raise ValueError("as-of must be a non-empty ISO-8601 timestamp")
normalized = value.strip()
if normalized.endswith(("Z", "z")):
normalized = normalized[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(normalized)
except ValueError as exc:
raise ValueError(f"Invalid ISO-8601 as-of timestamp: {value!r}") from exc
return require_aware(parsed, name="as-of")
def _provider_calendar(venue: str) -> tuple[str, str, Any]:
normalized, provider_name, timezone_name = _venue_contract(venue)
try:
import pandas_market_calendars as market_calendars
except (ImportError, TypeError) as exc:
raise CalendarUnavailableError(
"pandas-market-calendars==5.2.2 is required; install the skill's requirements.txt"
) from exc
try:
calendar = market_calendars.get_calendar(provider_name)
except Exception as exc:
raise CalendarUnavailableError(
f"Could not load exchange calendar for {normalized} ({provider_name}): {exc}"
) from exc
return normalized, timezone_name, calendar
def _schedule(venue: str, start_date: date, end_date: date):
normalized, timezone_name, calendar = _provider_calendar(venue)
try:
schedule = calendar.schedule(
start_date=start_date,
end_date=end_date,
tz=timezone_name,
market_times="all",
)
except Exception as exc:
raise CalendarUnavailableError(
f"Could not query exchange calendar for {normalized} "
f"from {start_date} through {end_date}: {exc}"
) from exc
return normalized, timezone_name, schedule
def _as_datetime(value: Any, *, name: str) -> Optional[datetime]:
if value is None or str(value) == "NaT":
return None
converter = getattr(value, "to_pydatetime", None)
converted = converter() if converter is not None else value
if not isinstance(converted, datetime):
raise CalendarUnavailableError(f"Calendar returned invalid {name}: {value!r}")
return require_aware(converted, name=name)
def _session_from_row(
venue: str,
timezone_name: str,
session_date: date,
row: Any,
) -> Session:
market_open = _as_datetime(row.get("market_open"), name="market_open")
market_close = _as_datetime(row.get("market_close"), name="market_close")
if market_open is None or market_close is None:
raise CalendarUnavailableError(
f"Calendar returned an incomplete session for {venue} on {session_date}"
)
break_start = _as_datetime(row.get("break_start"), name="break_start")
break_end = _as_datetime(row.get("break_end"), name="break_end")
if (break_start is None) != (break_end is None):
raise CalendarUnavailableError(
f"Calendar returned an incomplete break for {venue} on {session_date}"
)
return Session(
venue=venue,
session_date=session_date,
timezone=timezone_name,
market_open=market_open,
market_close=market_close,
break_start=break_start,
break_end=break_end,
)
def session_for_date(venue: str, session_date: date) -> Optional[Session]:
"""Return a session, or ``None`` for a valid holiday/weekend."""
session_date = _require_date(session_date, name="session_date")
normalized, timezone_name, schedule = _schedule(venue, session_date, session_date)
if schedule.empty:
return None
return _session_from_row(normalized, timezone_name, session_date, schedule.iloc[0])
def is_open_at(venue: str, timestamp: datetime) -> bool:
"""Return whether ``timestamp`` is inside regular trading hours."""
timestamp = require_aware(timestamp)
normalized, _, timezone_name = _venue_contract(venue)
local = timestamp.astimezone(ZoneInfo(timezone_name))
session = session_for_date(normalized, local.date())
if session is None:
return False
if not (session.market_open <= local < session.market_close):
return False
if session.break_start is not None and session.break_end is not None:
if session.break_start <= local < session.break_end:
return False
return True
def next_session_on_or_after(
venue: str,
start_date: date,
*,
max_search_days: int = 370,
) -> Session:
"""Return the first session on/after ``start_date`` within a finite horizon."""
start_date = _require_date(start_date, name="start_date")
if not isinstance(max_search_days, int) or isinstance(max_search_days, bool):
raise TypeError("max_search_days must be an integer")
if max_search_days < 0:
raise ValueError("max_search_days must be non-negative")
end_date = start_date + timedelta(days=max_search_days)
normalized, timezone_name, schedule = _schedule(venue, start_date, end_date)
if schedule.empty:
raise CalendarUnavailableError(
f"No {normalized} session found within {max_search_days} days of {start_date}"
)
label = schedule.index[0]
label_date = label.date() if hasattr(label, "date") else date.fromisoformat(str(label)[:10])
return _session_from_row(normalized, timezone_name, label_date, schedule.iloc[0])
def next_session_after(
venue: str,
session_date: date,
*,
max_search_days: int = 370,
) -> Session:
"""Return the first session strictly after ``session_date``."""
session_date = _require_date(session_date, name="session_date")
if max_search_days <= 0:
raise ValueError("max_search_days must be positive")
return next_session_on_or_after(
venue,
session_date + timedelta(days=1),
max_search_days=max_search_days - 1,
)
def previous_session_on_or_before(
venue: str,
end_date: date,
*,
max_search_days: int = 370,
) -> Session:
"""Return the last session on/before ``end_date`` within a finite horizon."""
end_date = _require_date(end_date, name="end_date")
if not isinstance(max_search_days, int) or isinstance(max_search_days, bool):
raise TypeError("max_search_days must be an integer")
if max_search_days < 0:
raise ValueError("max_search_days must be non-negative")
start_date = end_date - timedelta(days=max_search_days)
normalized, timezone_name, schedule = _schedule(venue, start_date, end_date)
if schedule.empty:
raise CalendarUnavailableError(
f"No {normalized} session found within {max_search_days} days of {end_date}"
)
label = schedule.index[-1]
label_date = label.date() if hasattr(label, "date") else date.fromisoformat(str(label)[:10])
return _session_from_row(normalized, timezone_name, label_date, schedule.iloc[-1])
def previous_session_before(
venue: str,
session_date: date,
*,
max_search_days: int = 370,
) -> Session:
"""Return the last session strictly before ``session_date``."""
session_date = _require_date(session_date, name="session_date")
if not isinstance(max_search_days, int) or isinstance(max_search_days, bool):
raise TypeError("max_search_days must be an integer")
if max_search_days <= 0:
raise ValueError("max_search_days must be positive")
return previous_session_on_or_before(
venue,
session_date - timedelta(days=1),
max_search_days=max_search_days - 1,
)
def current_or_next_session(venue: str, timestamp: datetime) -> Session:
"""Return today's not-yet-closed session, otherwise the next session."""
timestamp = require_aware(timestamp)
normalized, _, timezone_name = _venue_contract(venue)
local = timestamp.astimezone(ZoneInfo(timezone_name))
session = session_for_date(normalized, local.date())
if session is not None and local < session.market_close:
return session
return next_session_after(normalized, local.date())
def count_sessions(
venue: str,
start_date: date,
end_date: date,
*,
include_start: bool,
include_end: bool,
) -> int:
"""Count sessions with explicit date-boundary inclusion semantics."""
start_date = _require_date(start_date, name="start_date")
end_date = _require_date(end_date, name="end_date")
if start_date > end_date:
raise ValueError("start_date must be on or before end_date")
first = start_date if include_start else start_date + timedelta(days=1)
last = end_date if include_end else end_date - timedelta(days=1)
if first > last:
return 0
_, _, schedule = _schedule(venue, first, last)
return int(len(schedule.index))
scripts/breadth_csv_client.py
#!/usr/bin/env python3
"""
Breadth CSV Client - Fetch 200DMA breadth from TraderMonty's public CSV.
Lightweight client for auto-fetching S&P 500 breadth data (% above 200DMA)
from the market-breadth-analysis GitHub Pages CSV. No API key required.
Data source:
https://tradermonty.github.io/market-breadth-analysis/market_breadth_data.csv
"""
import csv
import io
from datetime import date, datetime
from typing import Optional
from zoneinfo import ZoneInfo
import requests
DEFAULT_DETAIL_URL = "https://tradermonty.github.io/market-breadth-analysis/market_breadth_data.csv"
TIMEOUT = 30
def fetch_breadth_200dma(
url: str = DEFAULT_DETAIL_URL,
*,
as_of_date: Optional[date] = None,
) -> Optional[dict]:
"""
Fetch the latest 200DMA breadth value from TraderMonty CSV.
Returns:
Dict with keys: value (0-100%), date (str), is_fresh (bool),
days_old (int), source (str).
None if fetch fails or CSV is empty.
"""
rows = _fetch_detail_csv(url)
if as_of_date is not None:
rows = [row for row in rows if _row_on_or_before(row, as_of_date)]
if not rows:
return None
latest = rows[-1]
raw_value = latest["Breadth_Index_Raw"]
value = round(raw_value * 100, 2)
date_str = latest["Date"]
freshness = _check_data_freshness(date_str, as_of_date=as_of_date)
return {
"value": value,
"date": date_str,
"is_fresh": freshness["is_fresh"],
"days_old": freshness["days_old"],
"source": "TraderMonty CSV",
}
def _fetch_detail_csv(url: str) -> list:
"""Fetch and parse the detail CSV. Returns rows sorted by date ascending."""
try:
resp = requests.get(url, timeout=TIMEOUT)
resp.raise_for_status()
except (requests.RequestException, OSError, ValueError):
return []
reader = csv.DictReader(io.StringIO(resp.text))
rows = []
for raw_row in reader:
try:
row = _parse_detail_row(raw_row)
rows.append(row)
except (ValueError, KeyError):
continue
if not rows:
return []
rows.sort(key=lambda r: r["Date"])
return rows
def _row_on_or_before(row: dict, as_of_date: date) -> bool:
try:
row_date = date.fromisoformat(row["Date"])
except (KeyError, TypeError, ValueError):
return False
return row_date <= as_of_date
def _check_data_freshness(
date_str: str,
max_stale_days: int = 5,
*,
as_of_date: Optional[date] = None,
) -> dict:
"""Check if data date is within acceptable freshness window.
Uses business days so Friday data remains fresh on Monday.
Future dates return is_fresh=False, days_old=None.
"""
from utils import count_business_days
try:
data_date = datetime.strptime(date_str, "%Y-%m-%d").date()
except ValueError:
return {"is_fresh": False, "days_old": None}
today = as_of_date or datetime.now(ZoneInfo("America/New_York")).date()
# Reject future dates
if data_date > today:
return {"is_fresh": False, "days_old": None}
biz_days = count_business_days(data_date, today)
return {
"is_fresh": biz_days <= max_stale_days,
"days_old": biz_days,
}
def _parse_detail_row(raw: dict) -> dict:
"""Convert a raw CSV row dict to properly typed values."""
return {
"Date": raw["Date"].strip(),
"S&P500_Price": float(raw["S&P500_Price"]),
"Breadth_Index_Raw": float(raw["Breadth_Index_Raw"]),
"Breadth_Index_200MA": float(raw["Breadth_Index_200MA"]),
"Breadth_Index_8MA": float(raw["Breadth_Index_8MA"]),
"Breadth_200MA_Trend": int(raw["Breadth_200MA_Trend"]),
"Bearish_Signal": _parse_bool(raw["Bearish_Signal"]),
"Is_Peak": _parse_bool(raw["Is_Peak"]),
"Is_Trough": _parse_bool(raw["Is_Trough"]),
"Is_Trough_8MA_Below_04": _parse_bool(raw["Is_Trough_8MA_Below_04"]),
}
def _parse_bool(val: str) -> bool:
"""Parse boolean from CSV string."""
return val.strip().lower() in ("true", "1", "yes")
scripts/calculators/__init__.py
"""Market Top Detector Calculator Modules"""
from .breadth_calculator import calculate_breadth_divergence
from .defensive_rotation_calculator import calculate_defensive_rotation
from .distribution_day_calculator import calculate_distribution_days
from .index_technical_calculator import calculate_index_technical
from .leading_stock_calculator import calculate_leading_stock_health
from .math_utils import calc_ema, calc_sma
from .sentiment_calculator import calculate_sentiment
__all__ = [
"calculate_distribution_days",
"calculate_leading_stock_health",
"calculate_defensive_rotation",
"calculate_breadth_divergence",
"calculate_index_technical",
"calculate_sentiment",
"calc_ema",
"calc_sma",
]
scripts/calculators/breadth_calculator.py
#!/usr/bin/env python3
"""
Component 4: Market Breadth Divergence (Weight: 15%)
Data source: CLI arguments (collected via WebSearch before script execution)
- breadth_200dma: % of S&P 500 stocks above 200DMA
- breadth_50dma: % of S&P 500 stocks above 50DMA
Scoring logic:
When index is near highs (within -5% of 52-week high):
200DMA_above < 40% -> 100 (Critical divergence)
200DMA_above < 50% -> 80
200DMA_above < 60% -> 60
200DMA_above < 70% -> 35
200DMA_above >= 70% -> 10
When index is NOT near highs (> -5%):
Score is halved (breadth divergence less meaningful when index is correcting)
50DMA breadth is used as a supplementary signal.
"""
from typing import Optional
def calculate_breadth_divergence(
breadth_200dma: Optional[float],
breadth_50dma: Optional[float],
index_distance_from_high_pct: float,
) -> dict:
"""
Calculate market breadth divergence score.
Args:
breadth_200dma: % of S&P 500 stocks above 200DMA (0-100)
breadth_50dma: % of S&P 500 stocks above 50DMA (0-100)
index_distance_from_high_pct: S&P 500 distance from 52-week high (negative %)
Returns:
Dict with score (0-100), signal, details
"""
if breadth_200dma is None:
return {
"score": 50,
"signal": "NO DATA: Breadth data not provided (neutral default)",
"data_available": False,
"breadth_200dma": None,
"breadth_50dma": breadth_50dma,
"index_near_highs": False,
"divergence_detected": False,
}
index_near_highs = index_distance_from_high_pct >= -5.0
# Primary score based on 200DMA breadth
raw_score = _score_200dma_breadth(breadth_200dma)
# Halve score if index is NOT near highs
if not index_near_highs:
raw_score = raw_score * 0.5
# 50DMA breadth supplementary adjustment
if breadth_50dma is not None:
if breadth_50dma < 30 and index_near_highs:
raw_score = min(100, raw_score + 10) # Extra warning
elif breadth_50dma > 70:
raw_score = max(0, raw_score - 5) # Slight relief
score = round(min(100, max(0, raw_score)))
# Detect divergence
divergence_detected = index_near_highs and breadth_200dma < 60
if score >= 80:
signal = "CRITICAL: Severe breadth divergence - index at highs but breadth collapsing"
elif score >= 60:
signal = "WARNING: Significant breadth divergence"
elif score >= 35:
signal = "CAUTION: Mild breadth weakness"
elif divergence_detected:
signal = "WATCH: Some breadth deterioration despite index strength"
else:
signal = "HEALTHY: Breadth supports index level"
return {
"score": score,
"signal": signal,
"data_available": True,
"breadth_200dma": breadth_200dma,
"breadth_50dma": breadth_50dma,
"index_near_highs": index_near_highs,
"index_distance_from_high_pct": round(index_distance_from_high_pct, 1),
"divergence_detected": divergence_detected,
}
def _score_200dma_breadth(breadth_200dma: float) -> int:
"""Score based on % of stocks above 200DMA.
Calibrated thresholds (linearly interpolated):
< 40% -> 100 (Critical divergence)
40-50% -> 100 to 55 (Severe)
50-60% -> 55 to 30 (Moderate)
60-70% -> 30 to 5 (Mild - still relatively healthy)
>= 70% -> 5 (Healthy breadth)
Calibration note: 62.26% should score ~24 (healthy/green signal)
"""
if breadth_200dma < 40:
return 100
elif breadth_200dma < 50:
return round(100 - (breadth_200dma - 40) / 10 * 45)
elif breadth_200dma < 60:
return round(55 - (breadth_200dma - 50) / 10 * 25)
elif breadth_200dma < 70:
return round(30 - (breadth_200dma - 60) / 10 * 25)
else:
return 5
scripts/calculators/defensive_rotation_calculator.py
#!/usr/bin/env python3
"""
Component 3: Defensive Sector Rotation (Weight: 15%)
Compares defensive ETF performance vs offensive/growth ETF performance
using multi-period weighted analysis: 10d(0.2) + 20d(0.5) + 40d(0.3).
Defensive: XLU, XLP, XLV, VNQ
Offensive: XLK, XLC, XLY, QQQ
Scoring (defensive_return - offensive_return):
+5.0% or more -> 100 (Strong rotation into defensives)
+3.0% or more -> 80
+1.5% or more -> 60
+0.5% or more -> 40
+0.0% or more -> 20
Negative -> 0 (Growth leading = healthy)
Multi-period confirmation:
All periods defensive -> "confirmed" (max 100)
Some periods only -> "unconfirmed" (capped at 80)
"""
from typing import Optional
DEFENSIVE_ETFS = ["XLU", "XLP", "XLV", "VNQ"]
OFFENSIVE_ETFS = ["XLK", "XLC", "XLY", "QQQ"]
MULTI_PERIOD_WEIGHTS = {10: 0.2, 20: 0.5, 40: 0.3}
def _calc_return(symbol_hist: list[dict], days: int) -> Optional[float]:
if not symbol_hist or len(symbol_hist) < days + 1:
return None
recent = symbol_hist[0].get("close", symbol_hist[0].get("adjClose", 0))
past = symbol_hist[days].get("close", symbol_hist[days].get("adjClose", 0))
if past == 0:
return None
return (recent - past) / past * 100
def _compute_period_rotation(historical: dict[str, list[dict]], lookback: int) -> Optional[dict]:
"""Compute defensive vs offensive rotation for a single period."""
def_returns = []
off_returns = []
for symbol in DEFENSIVE_ETFS:
hist = historical.get(symbol, [])
ret = _calc_return(hist, lookback)
if ret is not None:
def_returns.append(ret)
for symbol in OFFENSIVE_ETFS:
hist = historical.get(symbol, [])
ret = _calc_return(hist, lookback)
if ret is not None:
off_returns.append(ret)
if not def_returns or not off_returns:
return None
def_avg = sum(def_returns) / len(def_returns)
off_avg = sum(off_returns) / len(off_returns)
relative = def_avg - off_avg
score = _score_rotation(relative)
return {
"lookback": lookback,
"defensive_avg": round(def_avg, 2),
"offensive_avg": round(off_avg, 2),
"relative": round(relative, 2),
"score": score,
"defensive_leading": relative > 0,
}
def calculate_defensive_rotation(historical: dict[str, list[dict]], lookback: int = 20) -> dict:
"""
Calculate defensive vs offensive sector rotation score using multi-period analysis.
Args:
historical: Dict of symbol -> list of daily OHLCV (most recent first, 50+ days)
lookback: Primary lookback period (default 20, used for backward compat)
Returns:
Dict with score (0-100), relative_performance, multi_period details
"""
# Track fetch success
all_etfs = DEFENSIVE_ETFS + OFFENSIVE_ETFS
total_attempted = len(all_etfs)
fetch_successes = 0
def_details = {}
off_details = {}
for symbol in DEFENSIVE_ETFS:
hist = historical.get(symbol, [])
ret = _calc_return(hist, lookback)
if ret is not None:
def_details[symbol] = round(ret, 2)
fetch_successes += 1
for symbol in OFFENSIVE_ETFS:
hist = historical.get(symbol, [])
ret = _calc_return(hist, lookback)
if ret is not None:
off_details[symbol] = round(ret, 2)
fetch_successes += 1
fetch_success_rate = fetch_successes / total_attempted if total_attempted > 0 else 0.0
if not def_details or not off_details:
return {
"score": 50,
"signal": "INSUFFICIENT DATA (neutral default)",
"data_available": False,
"relative_performance": 0,
"defensive_avg_return": 0,
"offensive_avg_return": 0,
"defensive_details": def_details,
"offensive_details": off_details,
"lookback_days": lookback,
"fetch_success_rate": round(fetch_success_rate, 2),
}
# Multi-period analysis
period_results = {}
weighted_score = 0.0
total_weight = 0.0
all_defensive = True
periods_computed = 0
for period, weight in MULTI_PERIOD_WEIGHTS.items():
result = _compute_period_rotation(historical, period)
if result is not None:
period_results[period] = result
weighted_score += result["score"] * weight
total_weight += weight
periods_computed += 1
if not result["defensive_leading"]:
all_defensive = False
else:
all_defensive = False
# Fall back to single-period if multi-period unavailable
if total_weight > 0:
raw_score = weighted_score / total_weight
else:
# Fallback: single period
def_avg = sum(def_details.values()) / len(def_details)
off_avg = sum(off_details.values()) / len(off_details)
relative = def_avg - off_avg
raw_score = _score_rotation(relative)
# Multi-period confirmation
if periods_computed >= 2 and all_defensive:
confirmation = "confirmed"
score = round(min(100, raw_score))
elif periods_computed >= 2:
confirmation = "unconfirmed"
score = round(min(80, raw_score))
else:
confirmation = "single_period"
score = round(min(100, raw_score))
score = max(0, score)
# Primary period relative performance (for backward compat)
primary = period_results.get(lookback)
if primary:
relative = primary["relative"]
def_avg = primary["defensive_avg"]
off_avg = primary["offensive_avg"]
else:
def_avg = sum(def_details.values()) / len(def_details)
off_avg = sum(off_details.values()) / len(off_details)
relative = def_avg - off_avg
if score >= 80:
signal = "CRITICAL: Strong defensive rotation"
elif score >= 60:
signal = "WARNING: Defensive outperformance"
elif score >= 40:
signal = "CAUTION: Mild defensive rotation"
elif score >= 20:
signal = "MIXED: Slight defensive tilt"
else:
signal = "HEALTHY: Growth leading"
return {
"score": score,
"signal": signal,
"data_available": fetch_success_rate >= 0.75,
"relative_performance": round(relative, 2),
"defensive_avg_return": round(def_avg, 2),
"offensive_avg_return": round(off_avg, 2),
"defensive_details": def_details,
"offensive_details": off_details,
"lookback_days": lookback,
"fetch_success_rate": round(fetch_success_rate, 2),
"confirmation": confirmation,
"multi_period": period_results,
}
def _score_rotation(relative: float) -> int:
"""Convert relative performance (defensive - offensive) to 0-100 score"""
if relative >= 5.0:
return 100
elif relative >= 3.0:
return round(80 + (relative - 3.0) / 2.0 * 20)
elif relative >= 1.5:
return round(60 + (relative - 1.5) / 1.5 * 20)
elif relative >= 0.5:
return round(40 + (relative - 0.5) / 1.0 * 20)
elif relative >= 0.0:
return round(20 + relative / 0.5 * 20)
else:
if relative >= -2.0:
return round(max(0, 20 + relative / 2.0 * 20))
return 0
scripts/calculators/distribution_day_calculator.py
#!/usr/bin/env python3
"""
Component 1: Distribution Day Count (Weight: 25%)
O'Neil's Distribution Day Rules:
- Distribution Day: Index drops >= 0.2% on higher volume than previous day
- Stalling Day: Volume increases but price gain < 0.1% (half weight)
- Days expire after 25 trading days
- Count the higher of S&P 500 or NASDAQ
Scoring:
6+ distribution days -> 100 (Critical)
5 days -> 90
4 days -> 75 (O'Neil's warning threshold)
3 days -> 55
2 days -> 30
1 day -> 15
0 days -> 0
"""
def calculate_distribution_days(sp500_history: list[dict], nasdaq_history: list[dict]) -> dict:
"""
Calculate distribution day count for S&P 500 and NASDAQ.
Args:
sp500_history: List of daily OHLCV dicts (most recent first), at least 30 days
nasdaq_history: List of daily OHLCV dicts (most recent first), at least 30 days
Returns:
Dict with score (0-100), distribution_days, stalling_days, details
"""
sp500_result = _count_distribution_days(sp500_history, "S&P 500")
nasdaq_result = _count_distribution_days(nasdaq_history, "NASDAQ")
# Use the higher (worse) effective count
sp500_effective = sp500_result["distribution_days"] + 0.5 * sp500_result["stalling_days"]
nasdaq_effective = nasdaq_result["distribution_days"] + 0.5 * nasdaq_result["stalling_days"]
if sp500_effective >= nasdaq_effective:
primary = sp500_result
primary_name = "S&P 500"
else:
primary = nasdaq_result
primary_name = "NASDAQ"
effective_count = max(sp500_effective, nasdaq_effective)
raw_score = _score_distribution_days(effective_count)
# Clustering analysis: check if distribution days are concentrated recently
primary_details = primary["details"]
clustering = _calculate_clustering_factor(primary_details)
clustering_applied = effective_count >= 2 and clustering["factor"] > 0.5
if clustering_applied:
score = min(100, round(raw_score * 1.15))
else:
score = raw_score
# Build signal description
if effective_count >= 5:
signal = "CRITICAL: Heavy distribution detected"
elif effective_count >= 4:
signal = "WARNING: O'Neil's threshold reached"
elif effective_count >= 3:
signal = "CAUTION: Moderate distribution"
elif effective_count >= 1:
signal = "MINOR: Some distribution present"
else:
signal = "CLEAR: No distribution"
return {
"score": score,
"raw_score": raw_score,
"effective_count": effective_count,
"signal": signal,
"clustering": clustering,
"clustering_applied": clustering_applied,
"primary_index": primary_name,
"sp500": {
"distribution_days": sp500_result["distribution_days"],
"stalling_days": sp500_result["stalling_days"],
"effective_count": sp500_effective,
"details": sp500_result["details"],
},
"nasdaq": {
"distribution_days": nasdaq_result["distribution_days"],
"stalling_days": nasdaq_result["stalling_days"],
"effective_count": nasdaq_effective,
"details": nasdaq_result["details"],
},
}
def _count_distribution_days(history: list[dict], index_name: str) -> dict:
"""Count distribution and stalling days in the last 25 trading days"""
if not history or len(history) < 2:
return {"distribution_days": 0, "stalling_days": 0, "details": []}
# We need at least 26 days to check 25 days of change
# history[0] = most recent, history[1] = day before, etc.
window = min(25, len(history) - 1) # 25 trading day window
distribution_days = 0
stalling_days = 0
details = []
for i in range(window):
today = history[i]
yesterday = history[i + 1]
today_close = today.get("close", today.get("adjClose", 0))
yesterday_close = yesterday.get("close", yesterday.get("adjClose", 0))
today_volume = today.get("volume", 0)
yesterday_volume = yesterday.get("volume", 0)
if yesterday_close == 0 or yesterday_volume == 0:
continue
pct_change = (today_close - yesterday_close) / yesterday_close * 100
volume_increase = today_volume > yesterday_volume
date = today.get("date", f"day-{i}")
# Distribution day: price drops >= 0.2% AND volume increases
if pct_change <= -0.2 and volume_increase:
distribution_days += 1
details.append(
{
"date": date,
"type": "distribution",
"pct_change": round(pct_change, 2),
"volume_change": round((today_volume / yesterday_volume - 1) * 100, 1),
"window_index": i,
}
)
# Stalling day: volume increases but price gain < 0.1%
elif volume_increase and 0 <= pct_change < 0.1:
stalling_days += 1
details.append(
{
"date": date,
"type": "stalling",
"pct_change": round(pct_change, 2),
"volume_change": round((today_volume / yesterday_volume - 1) * 100, 1),
"window_index": i,
}
)
return {
"distribution_days": distribution_days,
"stalling_days": stalling_days,
"details": details,
}
def _calculate_clustering_factor(details: list) -> dict:
"""
Calculate how clustered distribution/stalling events are in recent days.
Looks at window_index of events: index 0-4 = most recent 5 days.
Returns factor = recent_count / total_count (0.0-1.0).
"""
if not details:
return {"factor": 0.0, "recent_count": 0, "total_count": 0}
total_count = len(details)
recent_count = sum(1 for d in details if d.get("window_index", 99) < 5)
factor = recent_count / total_count if total_count > 0 else 0.0
return {
"factor": round(factor, 2),
"recent_count": recent_count,
"total_count": total_count,
}
def _score_distribution_days(effective_count: float) -> int:
"""Convert effective distribution day count to 0-100 score"""
if effective_count >= 6:
return 100
elif effective_count >= 5:
return 90
elif effective_count >= 4:
return 75
elif effective_count >= 3:
return 55
elif effective_count >= 2:
return 30
elif effective_count >= 1:
return 15
else:
return 0
scripts/calculators/index_technical_calculator.py
#!/usr/bin/env python3
"""
Component 5: Index Technical Condition (Weight: 15%)
Evaluates technical health of S&P 500 and NASDAQ.
Uses shared historical data from Component 1 (no additional API calls).
Checkpoints (per index, S&P 500 + NASDAQ):
1. Price < 21 EMA (short-term weakness) -> +8pt
2. Price < 50 EMA (medium-term breakdown) -> +12pt
3. 21 EMA < 50 EMA (MA bearish crossover) -> +10pt
4. Price < 200 SMA (long-term trend breakdown) -> +15pt
5. Failed rally pattern -> +10pt
6. Lower highs pattern -> +10pt
7. Gap down on volume -> +10pt
Max per index: ~75 points (not all conditions fire at once in early stages)
Final: average of both indices, scaled to 0-100
"""
from typing import Optional
from calculators.math_utils import calc_ema
def calculate_index_technical(
sp500_history: list[dict],
nasdaq_history: list[dict],
sp500_quote: Optional[dict] = None,
nasdaq_quote: Optional[dict] = None,
) -> dict:
"""
Calculate index technical condition score.
Args:
sp500_history: S&P 500 daily OHLCV (most recent first, 250+ days preferred)
nasdaq_history: NASDAQ daily OHLCV (most recent first, 250+ days preferred)
sp500_quote: Current S&P 500 quote (optional, for real-time price)
nasdaq_quote: Current NASDAQ/QQQ quote (optional)
Returns:
Dict with score (0-100), sp500_details, nasdaq_details, signal
"""
sp500_result = _evaluate_index("S&P 500", sp500_history, sp500_quote)
nasdaq_result = _evaluate_index("NASDAQ", nasdaq_history, nasdaq_quote)
# Average only indices with available data
scores = []
if sp500_result.get("data_available", False):
scores.append(sp500_result["raw_score"])
if nasdaq_result.get("data_available", False):
scores.append(nasdaq_result["raw_score"])
if not scores:
final_score = 50 # Neutral when no data
signal = "NO DATA: Index data unavailable"
return {
"score": final_score,
"signal": signal,
"sp500": sp500_result,
"nasdaq": nasdaq_result,
"data_available": False,
}
avg_score = sum(scores) / len(scores)
final_score = round(min(100, max(0, avg_score)))
if final_score >= 70:
signal = "CRITICAL: Major technical breakdown"
elif final_score >= 50:
signal = "WARNING: Significant technical deterioration"
elif final_score >= 35:
signal = "CAUTION: Short-term weakness detected"
elif final_score >= 15:
signal = "MIXED: Minor technical concerns"
else:
signal = "HEALTHY: Technical structure intact"
return {
"score": final_score,
"signal": signal,
"sp500": sp500_result,
"nasdaq": nasdaq_result,
"data_available": True,
}
def _evaluate_index(name: str, history: list[dict], quote: Optional[dict] = None) -> dict:
"""Evaluate a single index's technical condition"""
if not history or len(history) < 21:
return {"raw_score": 0, "flags": ["Insufficient data"], "mas": {}, "data_available": False}
closes = [d.get("close", d.get("adjClose", 0)) for d in history]
highs = [d.get("high", d.get("close", 0)) for d in history]
[d.get("low", d.get("close", 0)) for d in history]
volumes = [d.get("volume", 0) for d in history]
# Current price (from quote or most recent close)
if quote:
price = quote.get("price", closes[0])
else:
price = closes[0]
score = 0
flags = []
mas = {}
# Calculate moving averages
ema21 = calc_ema(closes, 21)
mas["ema21"] = round(ema21, 2)
if len(closes) >= 50:
ema50 = calc_ema(closes, 50)
mas["ema50"] = round(ema50, 2)
else:
ema50 = None
if len(closes) >= 200:
sma200 = sum(closes[:200]) / 200
mas["sma200"] = round(sma200, 2)
else:
sma200 = None
# Check 1: Price < 21 EMA (short-term weakness)
if price < ema21:
score += 8
pct_below = (price - ema21) / ema21 * 100
flags.append(f"Below 21 EMA ({pct_below:+.1f}%)")
# Check 2: Price < 50 EMA (medium-term breakdown)
if ema50 and price < ema50:
score += 12
pct_below = (price - ema50) / ema50 * 100
flags.append(f"Below 50 EMA ({pct_below:+.1f}%)")
# Check 3: 21 EMA < 50 EMA (bearish crossover)
if ema50 and ema21 < ema50:
score += 10
flags.append("21 EMA < 50 EMA (bearish crossover)")
# Check 4: Price < 200 SMA (long-term trend breakdown)
if sma200 and price < sma200:
score += 15
pct_below = (price - sma200) / sma200 * 100
flags.append(f"Below 200 SMA ({pct_below:+.1f}%)")
# Check 5: Failed rally pattern
if _detect_failed_rally(closes, volumes):
score += 10
flags.append("Failed rally pattern detected")
# Check 6: Lower highs pattern (20 day)
if _detect_lower_highs(highs, lookback=20):
score += 10
flags.append("Lower highs pattern (20 day)")
# Check 7: Gap down on volume
if _detect_gap_down(history):
score += 10
flags.append("Recent gap down on volume")
return {
"raw_score": min(100, score),
"price": round(price, 2),
"flags": flags,
"mas": mas,
"data_available": True,
}
def _detect_failed_rally(closes: list[float], volumes: list[int], lookback: int = 15) -> bool:
"""
Detect failed rally: price bounces then fails to make new high
within recent lookback period.
"""
if len(closes) < lookback:
return False
recent = closes[:lookback]
# Find the peak in recent data
peak_idx = recent.index(max(recent))
# Failed rally: peak occurred 3-10 days ago, current price is below that peak
if 3 <= peak_idx <= 10:
# Price fell from peak
drop = (recent[0] - recent[peak_idx]) / recent[peak_idx] * 100
if drop < -2.0: # Dropped >2% from recent peak
return True
return False
def _detect_lower_highs(highs: list[float], lookback: int = 20) -> bool:
"""Detect lower highs pattern in price action"""
if len(highs) < lookback:
return False
recent_highs = highs[:lookback]
# Find swing highs (local maxima)
swing_highs = []
for i in range(1, len(recent_highs) - 1):
if recent_highs[i] > recent_highs[i - 1] and recent_highs[i] > recent_highs[i + 1]:
swing_highs.append(recent_highs[i])
if len(swing_highs) < 2:
return False
# Most recent swing high lower than previous
return swing_highs[0] < swing_highs[1]
def _detect_gap_down(history: list[dict], lookback: int = 5) -> bool:
"""Detect gap down on increased volume in recent sessions"""
if len(history) < lookback + 1:
return False
for i in range(lookback):
today = history[i]
yesterday = history[i + 1]
today_open = today.get("open", 0)
yesterday_close = yesterday.get("close", yesterday.get("adjClose", 0))
today_volume = today.get("volume", 0)
yesterday_volume = yesterday.get("volume", 0)
if yesterday_close == 0 or yesterday_volume == 0:
continue
gap_pct = (today_open - yesterday_close) / yesterday_close * 100
volume_increase = today_volume > yesterday_volume * 1.1
# Gap down >= 0.5% on higher volume
if gap_pct <= -0.5 and volume_increase:
return True
return False
scripts/calculators/leading_stock_calculator.py
#!/usr/bin/env python3
"""
Component 2: Leading Stock Health (Weight: 20%)
Evaluates health of leading/growth ETFs as proxy for market leadership.
Uses ETF baskets instead of individual stocks for API efficiency.
Default ETF Basket: ARKK, WCLD, IGV, XBI, SOXX, SMH, KWEB, TAN
Dynamic mode: Selects top-N ETFs from a 20-ETF candidate pool by 52-week high proximity.
Evaluation per ETF:
- Distance from 52-week high
- Position vs 50DMA and 200DMA
- Lower highs pattern detection
Scoring: Weighted average of ETF deterioration signals.
If 60%+ ETFs are deteriorating, apply 1.3x amplification.
"""
from typing import Optional
# Default Leading/Growth ETF basket
DEFAULT_LEADING_ETFS = ["ARKK", "WCLD", "IGV", "XBI", "SOXX", "SMH", "KWEB", "TAN"]
LEADING_ETFS = DEFAULT_LEADING_ETFS # backward compat
# Expanded candidate pool for dynamic basket selection
CANDIDATE_POOL = [
"SMH",
"SOXX",
"PSI",
"SOXQ", # Semiconductors
"IGV",
"WCLD",
"CLOU",
"BUG", # Software / Cloud / Cyber
"XBI",
"ARKG", # Biotech / Genomics
"ARKK",
"ARKW",
"KOMP", # Innovation / Disruptive
"TAN",
"ICLN", # Clean Energy
"KWEB",
"FDN", # Internet / China Tech
"FINX",
"IPAY", # Fintech / Payments
"BOTZ", # Robotics / AI
]
def select_dynamic_basket(quotes: dict[str, dict], top_n: int = 10) -> list[str]:
"""
Select top-N ETFs from candidate pool by proximity to 52-week high.
Args:
quotes: Dict of symbol -> quote data with 'price' and 'yearHigh'
top_n: Number of ETFs to select (default 10)
Returns:
List of selected symbols, sorted by proximity to high (closest first).
Falls back to DEFAULT_LEADING_ETFS if fewer than 5 valid candidates.
"""
scored = []
for symbol in CANDIDATE_POOL:
q = quotes.get(symbol)
if not q:
continue
price = q.get("price", 0)
year_high = q.get("yearHigh", 0)
if year_high <= 0 or price <= 0:
continue
proximity = price / year_high # 1.0 = at high, 0.5 = 50% below
scored.append((symbol, proximity))
if len(scored) < 5:
return list(DEFAULT_LEADING_ETFS)
# Sort by proximity descending (closest to high = strongest leader)
scored.sort(key=lambda x: x[1], reverse=True)
return [s[0] for s in scored[:top_n]]
def calculate_leading_stock_health(
quotes: dict[str, dict], historical: dict[str, list[dict]], etf_list: Optional[list[str]] = None
) -> dict:
"""
Calculate leading stock health score.
Args:
quotes: Dict of symbol -> quote data (from FMP batch quote)
historical: Dict of symbol -> list of daily OHLCV (most recent first, ~60 days)
etf_list: Optional explicit list of ETF symbols to evaluate.
If None, uses quotes.keys() or DEFAULT_LEADING_ETFS.
Returns:
Dict with score (0-100), etf_details, signal, basket_mode, basket
"""
etf_scores = []
etf_details = {}
if etf_list is not None:
basket_mode = "dynamic" if etf_list != list(DEFAULT_LEADING_ETFS) else "static"
eval_list = list(etf_list)
elif quotes:
basket_mode = "static"
eval_list = list(quotes.keys())
else:
basket_mode = "static"
eval_list = list(DEFAULT_LEADING_ETFS)
total_attempted = len(eval_list)
fetch_successes = 0
partial_data_count = 0
# Minimum history bars for full analysis (50DMA calculation)
MIN_HISTORY_BARS = 50
for symbol in eval_list:
quote = quotes.get(symbol)
hist = historical.get(symbol, [])
if not quote:
continue
# Full success requires quote + sufficient history
if len(hist) >= MIN_HISTORY_BARS:
fetch_successes += 1
else:
partial_data_count += 1
detail = _evaluate_etf(symbol, quote, hist)
etf_scores.append(detail["deterioration_score"])
etf_details[symbol] = detail
fetch_success_rate = fetch_successes / total_attempted if total_attempted > 0 else 0.0
if not etf_scores:
return {
"score": 50,
"signal": "INSUFFICIENT DATA",
"etf_details": {},
"etfs_evaluated": 0,
"etfs_deteriorating": 0,
"data_available": False,
"fetch_success_rate": fetch_success_rate,
"partial_data_count": partial_data_count,
"basket_mode": basket_mode,
"basket": eval_list,
}
# Mark data as unavailable if fetch success rate is below 75%
data_available = fetch_success_rate >= 0.75
avg_deterioration = sum(etf_scores) / len(etf_scores)
# Count deteriorating ETFs (score >= 50)
deteriorating_count = sum(1 for s in etf_scores if s >= 50)
deteriorating_pct = deteriorating_count / len(etf_scores)
# Amplification: if 60%+ ETFs deteriorating, multiply by 1.3
if deteriorating_pct >= 0.60:
final_score = min(100, avg_deterioration * 1.3)
else:
final_score = avg_deterioration
final_score = round(min(100, max(0, final_score)))
# Signal
if final_score >= 70:
signal = "CRITICAL: Leadership broadly deteriorating"
elif final_score >= 50:
signal = "WARNING: Multiple leaders weakening"
elif final_score >= 30:
signal = "CAUTION: Some leaders showing strain"
else:
signal = "HEALTHY: Leadership intact"
return {
"score": final_score,
"signal": signal,
"avg_deterioration": round(avg_deterioration, 1),
"etfs_evaluated": len(etf_scores),
"etfs_deteriorating": deteriorating_count,
"deteriorating_pct": round(deteriorating_pct * 100, 1),
"amplified": deteriorating_pct >= 0.60,
"etf_details": etf_details,
"data_available": data_available,
"fetch_success_rate": round(fetch_success_rate, 2),
"partial_data_count": partial_data_count,
"basket_mode": basket_mode,
"basket": eval_list,
}
def _evaluate_etf(symbol: str, quote: dict, history: list[dict]) -> dict:
"""Evaluate a single ETF's deterioration level"""
score = 0
flags = []
price = quote.get("price", 0)
year_high = quote.get("yearHigh", 0)
quote.get("yearLow", 0)
# 1. Distance from 52-week high (0-40 points)
if year_high > 0:
distance_pct = (price - year_high) / year_high * 100
if distance_pct <= -25:
score += 40
flags.append(f">{abs(distance_pct):.0f}% below 52wk high (bear territory)")
elif distance_pct <= -15:
score += 30
flags.append(f"{abs(distance_pct):.0f}% below 52wk high (correction)")
elif distance_pct <= -10:
score += 20
flags.append(f"{abs(distance_pct):.0f}% below 52wk high")
elif distance_pct <= -5:
score += 10
flags.append(f"{abs(distance_pct):.0f}% below 52wk high")
else:
distance_pct = 0
# 2. Position vs moving averages (0-40 points)
if history and len(history) >= 50:
closes = [d.get("close", d.get("adjClose", 0)) for d in history]
# 50DMA
sma50 = sum(closes[:50]) / 50
if price < sma50:
score += 20
flags.append(f"Below 50DMA (${sma50:.2f})")
# 200DMA (if enough data)
if len(closes) >= 200:
sma200 = sum(closes[:200]) / 200
if price < sma200:
score += 20
flags.append(f"Below 200DMA (${sma200:.2f})")
elif len(closes) >= 50:
# Estimate 200DMA position from available data
if price < sma50 * 0.95: # >5% below 50DMA suggests below 200DMA
score += 10
flags.append("Likely below 200DMA (estimated)")
elif history and len(history) >= 20:
closes = [d.get("close", d.get("adjClose", 0)) for d in history]
sma20 = sum(closes[:20]) / 20
if price < sma20:
score += 15
flags.append(f"Below 20DMA (${sma20:.2f})")
# 3. Lower highs pattern (0-20 points)
if history and len(history) >= 20:
lower_highs = _detect_lower_highs(history)
if lower_highs:
score += 20
flags.append("Lower highs pattern detected")
score = min(100, score)
return {
"deterioration_score": score,
"price": price,
"year_high": year_high,
"distance_from_high_pct": round(distance_pct, 1),
"flags": flags,
}
def _detect_lower_highs(history: list[dict], lookback: int = 20) -> bool:
"""
Detect lower highs pattern in recent price action.
Look for at least 2 consecutive lower swing highs in the last 20 days.
"""
if len(history) < lookback:
return False
highs = [d.get("high", d.get("close", 0)) for d in history[:lookback]]
# Find local maxima (swing highs)
swing_highs = []
for i in range(1, len(highs) - 1):
if highs[i] > highs[i - 1] and highs[i] > highs[i + 1]:
swing_highs.append(highs[i])
# Need at least 2 swing highs to compare
if len(swing_highs) < 2:
return False
# Check if most recent swing highs are declining
# swing_highs[0] is earliest (since history is most-recent-first, reversed)
# Actually history[0] = most recent, so highs[0] = most recent high
# So swing_highs are in reverse chronological order
return swing_highs[0] < swing_highs[1]
scripts/calculators/math_utils.py
#!/usr/bin/env python3
"""
Shared EMA/SMA calculation utilities for Market Top Detector.
All functions expect prices in most-recent-first order.
"""
def calc_ema(prices: list[float], period: int) -> float:
"""
Calculate Exponential Moving Average from prices (most recent first).
Args:
prices: List of prices, most recent first.
period: EMA period (must be >= 1).
Returns:
EMA value as float.
Raises:
ValueError: If prices is empty or period < 1.
"""
if not prices:
raise ValueError("prices must not be empty")
if period < 1:
raise ValueError("period must be >= 1")
if len(prices) < period:
return sum(prices) / len(prices)
prices_rev = prices[::-1]
sma = sum(prices_rev[:period]) / period
ema = sma
k = 2 / (period + 1)
for p in prices_rev[period:]:
ema = p * k + ema * (1 - k)
return ema
def calc_sma(prices: list[float], period: int) -> float:
"""
Calculate Simple Moving Average from prices (most recent first).
Args:
prices: List of prices, most recent first.
period: SMA period (must be >= 1).
Returns:
SMA value as float.
Raises:
ValueError: If prices is empty or period < 1.
"""
if not prices:
raise ValueError("prices must not be empty")
if period < 1:
raise ValueError("period must be >= 1")
if len(prices) < period:
return sum(prices) / len(prices)
return sum(prices[:period]) / period
scripts/calculators/sentiment_calculator.py
#!/usr/bin/env python3
"""
Component 6: Sentiment & Speculation (Weight: 10%)
Data sources:
- VIX: FMP API quote
- Put/Call ratio: CLI argument (from WebSearch)
- VIX term structure: CLI argument (from WebSearch) or auto-detected
- Margin Debt YoY: CLI argument (from WebSearch)
Scoring (additive, max 100):
Put/Call < 0.60 -> +35pt (extreme call buying = complacency)
Put/Call < 0.70 -> +25pt
Put/Call < 0.80 -> +12pt
Put/Call >= 0.80 -> +0pt (healthy caution)
VIX < 12 -> +25pt (extreme low fear)
VIX < 14 -> +17pt
VIX < 16 -> +8pt
VIX 16-20 -> +0pt (normal)
VIX > 25 -> -8pt (fear already present, top less likely)
VIX term structure:
Steep contango (normal) -> +25pt (complacency)
Normal contango -> +12pt
Flat -> +0pt
Backwardation -> -8pt (hedging demand = fear)
Margin Debt YoY:
> 30% -> +15pt (extreme leverage)
> 20% -> +10pt
> 10% -> +5pt
<= 10% -> +0pt
"""
from typing import Optional
# VIX term structure states
STEEP_CONTANGO = "steep_contango"
NORMAL_CONTANGO = "contango"
FLAT = "flat"
BACKWARDATION = "backwardation"
def calculate_sentiment(
vix_level: Optional[float] = None,
put_call_ratio: Optional[float] = None,
vix_term_structure: Optional[str] = None,
margin_debt_yoy_pct: Optional[float] = None,
) -> dict:
"""
Calculate sentiment and speculation score.
Args:
vix_level: Current VIX value (from FMP API)
put_call_ratio: CBOE equity put/call ratio (from WebSearch/CLI)
vix_term_structure: One of 'steep_contango', 'contango', 'flat', 'backwardation'
margin_debt_yoy_pct: Year-over-year margin debt change % (optional extra context)
Returns:
Dict with score (0-100), signal, component details
"""
# All inputs missing -> neutral default
if vix_level is None and put_call_ratio is None and vix_term_structure is None:
return {
"score": 50,
"signal": "NO DATA: All sentiment inputs missing (neutral default)",
"data_available": False,
"total_points": 0,
"details": {},
}
total_points = 0
details = {}
# 1. Put/Call Ratio scoring (max 35pt)
pc_points = 0
if put_call_ratio is not None:
if put_call_ratio < 0.60:
pc_points = 35
elif put_call_ratio < 0.70:
pc_points = 25
elif put_call_ratio < 0.80:
pc_points = 12
else:
pc_points = 0
details["put_call_ratio"] = {
"value": put_call_ratio,
"points": pc_points,
"interpretation": _interpret_put_call(put_call_ratio),
}
else:
details["put_call_ratio"] = {"value": None, "points": 0, "interpretation": "No data"}
total_points += pc_points
# 2. VIX Level scoring (max 25pt)
vix_points = 0
if vix_level is not None:
if vix_level < 12:
vix_points = 25
elif vix_level < 14:
vix_points = 17
elif vix_level < 16:
vix_points = 8
elif vix_level <= 25:
vix_points = 0
else:
vix_points = -8 # High fear = top less likely
details["vix_level"] = {
"value": round(vix_level, 2),
"points": vix_points,
"interpretation": _interpret_vix(vix_level),
}
else:
details["vix_level"] = {"value": None, "points": 0, "interpretation": "No data"}
total_points += vix_points
# 3. VIX Term Structure scoring (max 25pt)
vts_points = 0
if vix_term_structure:
if vix_term_structure == STEEP_CONTANGO:
vts_points = 25
elif vix_term_structure == NORMAL_CONTANGO:
vts_points = 12
elif vix_term_structure == FLAT:
vts_points = 0
elif vix_term_structure == BACKWARDATION:
vts_points = -8
details["vix_term_structure"] = {
"value": vix_term_structure,
"points": vts_points,
"interpretation": _interpret_vix_term(vix_term_structure),
}
else:
details["vix_term_structure"] = {"value": None, "points": 0, "interpretation": "No data"}
total_points += vts_points
# 4. Margin Debt YoY scoring (max 15pt)
md_points = 0
if margin_debt_yoy_pct is not None:
if margin_debt_yoy_pct > 30:
md_points = 15
elif margin_debt_yoy_pct > 20:
md_points = 10
elif margin_debt_yoy_pct > 10:
md_points = 5
else:
md_points = 0
details["margin_debt"] = {
"yoy_pct": margin_debt_yoy_pct,
"points": md_points,
"interpretation": _interpret_margin_debt(margin_debt_yoy_pct),
}
total_points += md_points
# Clamp to 0-100
score = max(0, min(100, total_points))
# Signal
if score >= 70:
signal = "CRITICAL: Extreme complacency / speculation"
elif score >= 50:
signal = "WARNING: Elevated complacency"
elif score >= 30:
signal = "CAUTION: Some speculative excess"
elif score >= 10:
signal = "MIXED: Moderate sentiment"
else:
signal = "HEALTHY: No excessive speculation"
return {
"score": score,
"signal": signal,
"data_available": True,
"total_points": total_points,
"details": details,
}
def _interpret_put_call(ratio: float) -> str:
if ratio < 0.60:
return "Extreme call buying - maximum complacency"
elif ratio < 0.70:
return "Elevated call buying - notable optimism"
elif ratio < 0.80:
return "Slightly optimistic"
elif ratio < 0.90:
return "Normal range"
else:
return "Elevated put buying - caution/fear present"
def _interpret_vix(level: float) -> str:
if level < 12:
return "Extreme low fear - complacency zone"
elif level < 14:
return "Low fear"
elif level < 16:
return "Mildly low volatility"
elif level <= 20:
return "Normal range"
elif level <= 25:
return "Elevated fear"
else:
return "High fear - panic present"
def _interpret_vix_term(structure: str) -> str:
mapping = {
STEEP_CONTANGO: "Market expects calm - complacency in term structure",
NORMAL_CONTANGO: "Normal term structure",
FLAT: "Uncertainty - no clear expectation",
BACKWARDATION: "Hedging demand elevated - fear in term structure",
}
return mapping.get(structure, "Unknown")
def _interpret_margin_debt(yoy_pct: float) -> str:
if yoy_pct >= 30:
return f"DANGER: Margin debt surging +{yoy_pct:.0f}% YoY"
elif yoy_pct >= 20:
return f"WARNING: Rapid leverage increase +{yoy_pct:.0f}% YoY"
elif yoy_pct >= 10:
return f"Elevated: +{yoy_pct:.0f}% YoY"
elif yoy_pct >= 0:
return f"Normal: +{yoy_pct:.0f}% YoY"
else:
return f"Deleveraging: {yoy_pct:.0f}% YoY"
scripts/fmp_client.py
#!/usr/bin/env python3
# GENERATED by scripts/generate_fmp_client.py — do not edit.
# Source of truth: scripts/fmp_client/specials/ and scripts/fmp_client/registry.py.
# Regenerate: python3 scripts/generate_fmp_client.py
"""
FMP API Client for Market Top Detector
Provides rate-limited access to Financial Modeling Prep API endpoints
for market top detection analysis.
Features:
- Rate limiting (0.3s between requests)
- Automatic retry on 429 errors
- Session caching for duplicate requests
- Batch quote support for ETF baskets
"""
import os
import re
import sys
import time
from datetime import datetime, timedelta
from typing import Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
try:
import requests
except ImportError:
print("ERROR: requests library not found. Install with: pip install requests", file=sys.stderr)
sys.exit(1)
# Mirror of scripts/provider_contracts.py::redact_url (clients are standalone).
# Requires a `?`/`&` prefix, so this masks apikey=/api_key= in URL query
# strings, not a rendered params dict (unreachable today: only
# RequestException text -- which can embed the full request URL -- is
# ever printed here, never a bare params repr()).
_APIKEY_RE = re.compile(r"([?&](?:apikey|api_key)=)[^&\s]+", re.IGNORECASE)
def _redact_key(text: str) -> str:
"""Mask apikey=/api_key= query values so stderr never carries the key."""
return _APIKEY_RE.sub(r"\1REDACTED", str(text))
# --- FMP endpoint fallback: stable (new users) -> v3 (legacy users) ---
def _et_today(now=None):
"""America/New_York calendar date for provider query windows (issue #427).
An explicit aware ``now`` wins (deterministic runs/tests); otherwise the
current instant is converted to America/New_York. ``now`` must be
timezone-aware when supplied. Fail-closed: missing TZ data raises
RuntimeError naming the tzdata requirement instead of silently falling
back to the runner's local clock.
"""
try:
tz = ZoneInfo("America/New_York")
except ZoneInfoNotFoundError as exc:
raise RuntimeError("America/New_York timezone data is missing; pip install tzdata") from exc
if now is not None:
if now.tzinfo is None:
raise ValueError("now must be timezone-aware")
return now.astimezone(tz).date()
return datetime.now(tz).date()
def _stable_quote_url(base, symbols_str, params):
"""stable/quote?symbol=^GSPC"""
params["symbol"] = symbols_str
return base, params
def _v3_quote_url(base, symbols_str, params):
"""api/v3/quote/^GSPC"""
return f"{base}/{symbols_str}", params
def _stable_hist_url(base, symbols_str, params):
"""stable/historical-price-eod/full?symbol=^GSPC&from=...&to=..."""
params["symbol"] = symbols_str
# New stable EOD endpoint ignores `timeseries`; convert to from/to range
# to bound the payload. Use 2x calendar days to cover N trading days
# (trading-day/calendar-day ratio ~252/365 ~0.69, so *2 leaves headroom).
days = params.pop("timeseries", None)
if days is not None:
today = _et_today()
params["from"] = (today - timedelta(days=int(days) * 2)).isoformat()
params["to"] = today.isoformat()
return base, params
def _v3_hist_url(base, symbols_str, params):
"""api/v3/historical-price-full/^GSPC?timeseries=80"""
return f"{base}/{symbols_str}", params
_FMP_ENDPOINTS = {
"quote": [
("https://financialmodelingprep.com/stable/quote", _stable_quote_url),
("https://financialmodelingprep.com/api/v3/quote", _v3_quote_url),
],
"historical": [
("https://financialmodelingprep.com/stable/historical-price-eod/full", _stable_hist_url),
("https://financialmodelingprep.com/api/v3/historical-price-full", _v3_hist_url),
],
}
def _normalize_eod_flat_list(data, symbols_str: str, limit: Optional[int] = None):
"""Convert stable/historical-price-eod/full flat list to v3-compatible dict.
Input : [{"symbol": "SPY", "date": "...", "open": ..., ...}, ...]
Output : {"symbol": "SPY", "historical": [{"date": ..., "open": ..., ...}, ...]}
Returns the input unchanged if not a list (passthrough for v3 dict /
historicalStockList responses). Returns None when no row matches the
requested symbol; the caller will record the failure and try the next
endpoint.
If `limit` is provided (the original `timeseries=N` request), the
`historical` list is truncated to the first `limit` entries. The new
EOD endpoint ignores `timeseries` and returns the full available history,
so the caller's date-range bounding plus this truncation together preserve
the legacy "most-recent N rows" contract. Truncation assumes descending
date order, which the FMP EOD endpoint provides (verified live).
Note: empty list ``[]`` does not reach this normalizer because the caller's
``if not data: continue`` falsy check handles it earlier in
``_request_with_fallback``.
"""
if not isinstance(data, list):
return data
if not data:
return None
norm_target = symbols_str.replace("-", ".")
matched_symbol = None
historical = []
for row in data:
if not isinstance(row, dict):
continue
# Be permissive: single-symbol endpoint may omit per-row "symbol".
# Treat missing symbol as belonging to the requested symbols_str.
row_sym = row.get("symbol") or symbols_str
if row_sym.replace("-", ".") != norm_target:
continue
matched_symbol = matched_symbol or row_sym
historical.append({k: v for k, v in row.items() if k != "symbol"})
if not historical:
return None
if limit is not None and limit > 0:
historical = historical[:limit]
return {"symbol": matched_symbol or symbols_str, "historical": historical}
def _has_usable_history(data) -> bool:
"""True only when `data` is a dict carrying a non-empty `historical` list.
A dict with an empty `historical` list (e.g. an index/ETF unavailable on the
caller's FMP plan) is treated as unusable so the caller can fall back to
yfinance instead of caching an empty result.
"""
return bool(
isinstance(data, dict) and isinstance(data.get("historical"), list) and data["historical"]
)
class FMPClient:
"""Client for Financial Modeling Prep API with rate limiting and caching"""
BASE_URL = "https://financialmodelingprep.com/api/v3"
RATE_LIMIT_DELAY = 0.3 # 300ms between requests
_ENDPOINT_FAILURE_THRESHOLD = 3 # disable endpoint after N consecutive failures
def _redact_secret(self, text) -> str:
"""Mask raw and URL-encoded key echoes before diagnostics are truncated."""
from urllib.parse import quote, quote_plus
text = _redact_key(text)
if not self.api_key:
return text
patterns = {re.escape(self.api_key)}
for encoded in (quote(self.api_key, safe=""), quote_plus(self.api_key, safe="")):
# Percent triplets are case-insensitive; the key's letters are not.
parts = re.split(r"(%[0-9A-Fa-f]{2})", encoded)
patterns.add(
"".join(
"(?i:" + re.escape(part) + ")"
if re.fullmatch(r"%[0-9A-Fa-f]{2}", part)
else re.escape(part)
for part in parts
)
)
# One pass prevents a short key from matching the replacement marker.
return re.sub("|".join(sorted(patterns, key=len, reverse=True)), "***REDACTED***", text)
def __init__(self, api_key: Optional[str] = None):
self.api_key = (api_key or os.getenv("FMP_API_KEY") or "").strip()
if not self.api_key:
raise ValueError(
"FMP API key required. Set FMP_API_KEY environment variable "
"or pass api_key parameter."
)
self.session = requests.Session()
self.session.headers.update({"apikey": self.api_key})
self.cache = {}
self.last_call_time = 0
self.rate_limit_reached = False
self.retry_count = 0
self.max_retries = 1
self.api_calls_made = 0
# Circuit breaker: track consecutive failures per endpoint URL prefix
self._endpoint_failures: dict[str, int] = {}
self._disabled_endpoints: set[str] = set()
def _rate_limited_get(
self, url: str, params: Optional[dict] = None, quiet: bool = False
) -> Optional[dict]:
if self.rate_limit_reached:
return None
if params is None:
params = {}
elapsed = time.time() - self.last_call_time
if elapsed < self.RATE_LIMIT_DELAY:
time.sleep(self.RATE_LIMIT_DELAY - elapsed)
try:
response = self.session.get(url, params=params, timeout=30)
self.last_call_time = time.time()
self.api_calls_made += 1
if response.status_code == 200:
self.retry_count = 0
return response.json()
elif response.status_code == 429:
self.retry_count += 1
if self.retry_count <= self.max_retries:
print("WARNING: Rate limit exceeded. Waiting 60 seconds...", file=sys.stderr)
time.sleep(60)
return self._rate_limited_get(url, params, quiet=quiet)
else:
print("ERROR: Daily API rate limit reached.", file=sys.stderr)
self.rate_limit_reached = True
return None
else:
if not quiet:
print(
self._redact_secret(
f"ERROR: API request failed: {response.status_code} - {self._redact_secret(response.text)[:200]}"
),
file=sys.stderr,
)
return None
except requests.exceptions.RequestException as e:
print(self._redact_secret(f"ERROR: Request exception: {e}"), file=sys.stderr)
return None
def _request_with_fallback(self, endpoint_key, symbols_str, extra_params=None):
"""Try stable endpoint first, fall back to v3 for legacy users.
Returns parsed JSON in v3-compatible shape, or None if all fail.
Non-last endpoints use quiet=True to suppress expected 403 stderr.
"""
params = dict(extra_params) if extra_params else {}
endpoints = _FMP_ENDPOINTS[endpoint_key]
is_single = "," not in symbols_str
for i, (base_url, url_builder) in enumerate(endpoints):
# Circuit breaker: skip endpoints with too many consecutive failures
if base_url in self._disabled_endpoints:
continue
url, final_params = url_builder(base_url, symbols_str, dict(params))
is_last = i == len(endpoints) - 1
data = self._rate_limited_get(url, final_params, quiet=not is_last)
if not data: # falsy (None, [], {}) — try next endpoint
self._record_endpoint_failure(base_url)
continue
# Normalize new stable EOD flat-list shape to v3-compatible dict.
# No-op for v3 dict / historicalStockList responses.
# `timeseries` (original request) is passed as `limit` so the
# EOD endpoint's full-history response is truncated to the
# legacy "most-recent N rows" contract.
if endpoint_key == "historical":
limit = params.get("timeseries") if isinstance(params, dict) else None
data = _normalize_eod_flat_list(data, symbols_str, limit=limit)
if not data:
self._record_endpoint_failure(base_url)
continue
# Shape validation: reject truthy-but-wrong-shape responses
valid = True
if endpoint_key == "quote":
if not isinstance(data, list) or len(data) == 0:
valid = False
elif is_single and not any(
q.get("symbol", "").replace("-", ".") == symbols_str.replace("-", ".")
for q in data
):
valid = False
if endpoint_key == "historical":
if not isinstance(data, dict):
valid = False
elif "historicalStockList" in data:
# stable batch format -> v3 single format (exact match only)
norm = symbols_str.replace("-", ".")
found = None
for entry in data["historicalStockList"]:
if entry.get("symbol", "").replace("-", ".") == norm:
found = {
"symbol": entry.get("symbol"),
"historical": entry.get("historical", []),
}
break
if found:
self._endpoint_failures[base_url] = 0
return found
valid = False
elif "historical" not in data:
valid = False
elif is_single and data.get("symbol"):
if data["symbol"].replace("-", ".") != symbols_str.replace("-", "."):
valid = False
if valid:
self._endpoint_failures[base_url] = 0
return data
self._record_endpoint_failure(base_url)
return None
def _record_endpoint_failure(self, base_url: str) -> None:
"""Track consecutive failures and disable endpoint after threshold."""
failures = self._endpoint_failures.get(base_url, 0) + 1
self._endpoint_failures[base_url] = failures
if failures >= self._ENDPOINT_FAILURE_THRESHOLD:
self._disabled_endpoints.add(base_url)
def get_quote(self, symbols: str) -> Optional[list[dict]]:
"""Fetch real-time quote data for one or more symbols (comma-separated).
Falls back to a yfinance-derived quote when FMP returns nothing (e.g. an
index like ^GSPC/^VIX gated behind a paid FMP plan). Fallback is
single-symbol only — comma lists are left to FMP.
"""
cache_key = f"quote_{symbols}"
if cache_key in self.cache:
return self.cache[cache_key]
data = self._request_with_fallback("quote", symbols)
if not data and "," not in symbols:
data = self._get_quote_from_yfinance(symbols)
if data:
self.cache[cache_key] = data
return data
def _get_quote_from_yfinance(self, symbol: str) -> Optional[list[dict]]:
"""Fallback quote derived from recent yfinance daily bars.
Returns the FMP quote shape (a list with one dict carrying
price/previousClose/change/changesPercentage/open/high/low/volume/
yearHigh/yearLow), or ``None`` when history is unavailable. Year
high/low are approximated from ~252 daily bars.
"""
hist = self._get_hist_from_yfinance(symbol, 260)
if not _has_usable_history(hist):
return None
bars = hist["historical"] # most-recent-first
latest = bars[0]
prev_close = bars[1]["close"] if len(bars) > 1 else latest["close"]
price = latest["close"]
change = price - prev_close
pct = (change / prev_close * 100) if prev_close else 0.0
window = bars[:252]
year_high = max(b["high"] for b in window)
year_low = min(b["low"] for b in window)
return [
{
"symbol": symbol,
"price": price,
"previousClose": prev_close,
"change": round(change, 4),
"changesPercentage": round(pct, 4),
"open": latest["open"],
"dayHigh": latest["high"],
"dayLow": latest["low"],
"high": latest["high"],
"low": latest["low"],
"volume": latest["volume"],
"yearHigh": year_high,
"yearLow": year_low,
}
]
def get_historical_prices(self, symbol: str, days: int = 365) -> Optional[dict]:
"""Fetch historical daily OHLCV data.
Falls back to yfinance when the FMP endpoint returns no usable history
(e.g. an index/ETF gated behind a paid FMP plan). The yfinance path
needs no extra API key; an FMP key is still required to build the client.
"""
cache_key = f"prices_{symbol}_{days}"
if cache_key in self.cache:
return self.cache[cache_key]
data = self._request_with_fallback("historical", symbol, {"timeseries": days})
if not _has_usable_history(data):
data = self._get_hist_from_yfinance(symbol, days)
if _has_usable_history(data):
self.cache[cache_key] = data
return data
return None
def _get_hist_from_yfinance(self, symbol: str, days: int) -> Optional[dict]:
"""Fallback: fetch daily history via yfinance when FMP is unavailable.
Returns the FMP-compatible shape ``{"symbol": ..., "historical": [...]}``
with most-recent-first bars (date/open/high/low/close/adjClose/volume),
or ``None`` on empty/error. ``yfinance`` is imported lazily so the FMP
success path never depends on it.
"""
try:
import yfinance as yf
end = _et_today()
start = end - timedelta(days=int(days * 1.5)) # cover weekends/holidays
df = yf.download(
symbol,
start=start.isoformat(),
end=end.isoformat(),
auto_adjust=True,
progress=False,
)
if df is None or df.empty:
return None
# yfinance returns MultiIndex columns for a single ticker.
if hasattr(df.columns, "levels"):
df.columns = df.columns.droplevel(1)
historical = []
for idx, row in df.iterrows():
close = float(row["Close"])
historical.append(
{
"date": idx.strftime("%Y-%m-%d"),
"open": float(row["Open"]),
"high": float(row["High"]),
"low": float(row["Low"]),
"close": close,
"adjClose": close, # auto_adjust=True -> Close already adjusted
"volume": int(row["Volume"]),
}
)
if not historical:
return None
historical.reverse() # yfinance ascending -> FMP most-recent-first
return {"symbol": symbol, "historical": historical[:days]}
except Exception as e:
print(
self._redact_secret(f"WARNING: yfinance history fallback failed for {symbol}: {e}"),
file=sys.stderr,
)
return None
def get_batch_quotes(self, symbols: list[str]) -> dict[str, dict]:
"""Fetch quotes for a list of symbols.
Note: The FMP `/stable/quote` endpoint only accepts a single symbol per
call. Passing comma-separated symbols silently returns []. The
deprecated `/api/v3/quote/A,B,C` batch syntax now returns 403 for
non-legacy keys (retired Aug 31, 2025). So we iterate one symbol at a
time. Rate limiting in `_rate_limited_get` keeps this within plan
limits.
"""
results = {}
for symbol in symbols:
quotes = self.get_quote(symbol)
if quotes:
for q in quotes:
if q.get("symbol"):
results[q["symbol"]] = q
return results
def get_batch_historical(self, symbols: list[str], days: int = 50) -> dict[str, list[dict]]:
"""Fetch historical prices for multiple symbols"""
results = {}
for symbol in symbols:
data = self.get_historical_prices(symbol, days=days)
if data and "historical" in data:
results[symbol] = data["historical"]
return results
def calculate_ema(self, prices: list[float], period: int) -> float:
"""Calculate EMA (thin wrapper around math_utils for backward compat)."""
from calculators.math_utils import calc_ema
return calc_ema(prices, period)
def calculate_sma(self, prices: list[float], period: int) -> float:
"""Calculate SMA (thin wrapper around math_utils for backward compat)."""
from calculators.math_utils import calc_sma
return calc_sma(prices, period)
def get_vix_term_structure(self) -> Optional[dict]:
"""
Auto-detect VIX term structure by comparing VIX to VIX3M.
Returns:
Dict with ratio, classification, or None if VIX3M unavailable.
"""
vix_quotes = self.get_quote("^VIX")
vix3m_quotes = self.get_quote("^VIX3M")
if not vix_quotes or not vix3m_quotes:
return None
vix_price = vix_quotes[0].get("price", 0)
vix3m_price = vix3m_quotes[0].get("price", 0)
if vix3m_price <= 0:
return None
ratio = vix_price / vix3m_price
if ratio < 0.85:
classification = "steep_contango"
elif ratio < 0.95:
classification = "contango"
elif ratio <= 1.05:
classification = "flat"
else:
classification = "backwardation"
return {
"vix": round(vix_price, 2),
"vix3m": round(vix3m_price, 2),
"ratio": round(ratio, 3),
"classification": classification,
}
def get_api_stats(self) -> dict:
return {
"cache_entries": len(self.cache),
"api_calls_made": self.api_calls_made,
"rate_limit_reached": self.rate_limit_reached,
}
scripts/historical_comparator.py
#!/usr/bin/env python3
"""
Historical Top Pattern Comparator
Compares current component scores to estimated score patterns from
historical market tops (2000, 2007, 2018, 2022) using Sum of Squared
Differences (SSD) to find the closest match.
"""
import math
from typing import Optional
# Predeclared heuristic tolerance, not fitted or calibrated on outcomes.
MAX_COMPARABLE_RMS = 15.0
# Component weights (must match scorer.py)
COMPONENT_WEIGHTS = {
"distribution_days": 0.25,
"leading_stocks": 0.20,
"defensive_rotation": 0.15,
"breadth_divergence": 0.15,
"index_technical": 0.15,
"sentiment": 0.10,
}
# Estimated median component scores during historical tops
# Based on references/historical_tops.md analysis
HISTORICAL_TOPS = {
"2000 (Dot-Com Bubble)": {
"distribution_days": 90,
"leading_stocks": 85,
"defensive_rotation": 70,
"breadth_divergence": 95,
"index_technical": 60,
"sentiment": 90,
},
"2007 (Financial Crisis)": {
"distribution_days": 80,
"leading_stocks": 75,
"defensive_rotation": 85,
"breadth_divergence": 80,
"index_technical": 70,
"sentiment": 75,
},
"2018 (Q4 Correction)": {
"distribution_days": 60,
"leading_stocks": 50,
"defensive_rotation": 40,
"breadth_divergence": 55,
"index_technical": 65,
"sentiment": 55,
},
"2022 (Rate Hike Bear)": {
"distribution_days": 75,
"leading_stocks": 90,
"defensive_rotation": 60,
"breadth_divergence": 70,
"index_technical": 80,
"sentiment": 65,
},
}
COMPONENT_KEYS = [
"distribution_days",
"leading_stocks",
"defensive_rotation",
"breadth_divergence",
"index_technical",
"sentiment",
]
def _compute_ssd(current: dict[str, float], historical: dict[str, float]) -> float:
"""Compute weighted sum of squared differences between two score dicts.
Each component's squared difference is multiplied by its weight,
aligning the distance metric with the composite score design.
"""
ssd = 0.0
for key in COMPONENT_KEYS:
diff = current.get(key, 0) - historical.get(key, 0)
weight = COMPONENT_WEIGHTS.get(key, 1.0 / len(COMPONENT_KEYS))
ssd += weight * diff * diff
return ssd
def compare_to_historical(
current_scores: dict[str, float], data_availability: Optional[dict[str, bool]] = None
) -> dict:
"""
Compare current scores to historical top patterns.
Args:
current_scores: Dict of component key -> score (0-100)
Returns:
Dict with closest_match, all_comparisons, narrative
"""
invalid = [
key
for key in COMPONENT_KEYS
if (data_availability is not None and not data_availability.get(key, True))
or isinstance(current_scores.get(key), bool)
or not isinstance(current_scores.get(key), (int, float))
or not math.isfinite(current_scores[key])
or not 0 <= current_scores[key] <= 100
]
if invalid:
return {
"status": "no_comparable_episode",
"closest_match": None,
"closest_ssd": None,
"nearest_candidate": None,
"nearest_rms": None,
"comparisons": [],
"max_comparable_rms": MAX_COMPARABLE_RMS,
"narrative": "No comparable episode: unavailable or invalid components: "
+ ", ".join(invalid)
+ ".",
}
comparisons = []
for name, hist_scores in HISTORICAL_TOPS.items():
ssd = _compute_ssd(current_scores, hist_scores)
comparisons.append(
{
"name": name,
"ssd": round(ssd, 1),
"rms": math.sqrt(ssd),
"scores": hist_scores,
}
)
comparisons.sort(key=lambda x: x["rms"])
closest = comparisons[0]
comparable = closest["rms"] <= MAX_COMPARABLE_RMS
narrative = (
_generate_narrative(current_scores, closest)
if comparable
else "No comparable episode: nearest illustrative pattern exceeds the heuristic "
f"{MAX_COMPARABLE_RMS:g}-point weighted RMS tolerance."
)
narrative += " Historical patterns are retrospective estimates, not calibration evidence."
return {
"status": "comparable_illustration" if comparable else "no_comparable_episode",
"closest_match": closest["name"] if comparable else None,
"closest_ssd": closest["ssd"] if comparable else None,
"nearest_candidate": closest["name"],
"nearest_rms": closest["rms"],
"max_comparable_rms": MAX_COMPARABLE_RMS,
"comparisons": comparisons,
"narrative": narrative,
}
def _generate_narrative(current_scores: dict[str, float], closest: dict) -> str:
"""Generate a textual comparison highlighting key differences."""
hist_scores = closest["scores"]
name = closest["name"]
diffs = []
for key in COMPONENT_KEYS:
curr = current_scores.get(key, 0)
hist = hist_scores.get(key, 0)
diff = curr - hist
if abs(diff) >= 15:
direction = "higher" if diff > 0 else "lower"
label = key.replace("_", " ").title()
diffs.append(f"{label} is {abs(diff):.0f}pt {direction} than {name}")
if not diffs:
return f"Current pattern closely matches {name} across all components."
diff_text = "; ".join(diffs)
return f"Closest to {name}. Key differences: {diff_text}."
scripts/market_top_detector.py
#!/usr/bin/env python3
"""
Market Top Detector - Main Orchestrator
Integrates O'Neil (Distribution Days), Minervini (Leading Stock Deterioration),
and Monty (Defensive Sector Rotation) approaches to assess heuristic market top risk.
Usage:
# With FMP API (recommended):
python3 market_top_detector.py --api-key YOUR_KEY \\
--breadth-200dma 62.26 --breadth-50dma 55.0 \\
--put-call 0.67 --vix-term contango
# Using environment variable:
export FMP_API_KEY=YOUR_KEY
python3 market_top_detector.py --breadth-200dma 62.26
# Minimal (VIX from API, rest from CLI):
python3 market_top_detector.py --api-key YOUR_KEY \\
--breadth-200dma 62.26 --put-call 0.67
Output:
- JSON: market_top_YYYY-MM-DD_HHMMSS.json
- Markdown: market_top_YYYY-MM-DD_HHMMSS.md
"""
import argparse
import glob
import json
import os
import sys
from datetime import date, datetime
from typing import Optional
from zoneinfo import ZoneInfo
# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(__file__))
from breadth_csv_client import fetch_breadth_200dma
from calculators.breadth_calculator import calculate_breadth_divergence
from calculators.defensive_rotation_calculator import (
DEFENSIVE_ETFS,
OFFENSIVE_ETFS,
calculate_defensive_rotation,
)
from calculators.distribution_day_calculator import calculate_distribution_days
from calculators.index_technical_calculator import calculate_index_technical
from calculators.leading_stock_calculator import (
CANDIDATE_POOL,
LEADING_ETFS,
calculate_leading_stock_health,
select_dynamic_basket,
)
from calculators.sentiment_calculator import calculate_sentiment
from fmp_client import FMPClient
from historical_comparator import compare_to_historical
from report_generator import generate_json_report, generate_markdown_report
from scenario_engine import generate_scenarios
from scorer import calculate_composite_score, detect_follow_through_day
def _iso_date(value: str) -> date:
try:
parsed = date.fromisoformat(value)
except ValueError as exc:
raise argparse.ArgumentTypeError("must be YYYY-MM-DD") from exc
if parsed.isoformat() != value:
raise argparse.ArgumentTypeError("must be zero-padded YYYY-MM-DD")
return parsed
def parse_arguments(argv: Optional[list[str]] = None):
parser = argparse.ArgumentParser(
description="Market Top Detector - O'Neil/Minervini/Monty Integration"
)
# API key
parser.add_argument(
"--api-key", help="FMP API key (defaults to FMP_API_KEY environment variable)"
)
# WebSearch-sourced data (provided by Claude before script execution)
parser.add_argument(
"--breadth-200dma",
type=float,
default=None,
help="Percent of S&P 500 stocks above 200DMA (e.g., 62.26)",
)
parser.add_argument(
"--breadth-50dma",
type=float,
default=None,
help="Percent of S&P 500 stocks above 50DMA (e.g., 55.0)",
)
parser.add_argument(
"--put-call", type=float, default=None, help="CBOE equity put/call ratio (e.g., 0.67)"
)
parser.add_argument(
"--vix-term",
choices=["steep_contango", "contango", "flat", "backwardation"],
default=None,
help="VIX term structure state",
)
parser.add_argument(
"--margin-debt-yoy",
type=float,
default=None,
help="Margin debt year-over-year change percent (e.g., 36.0)",
)
# Data freshness dates
parser.add_argument(
"--breadth-200dma-date", default=None, help="Date of breadth 200DMA data (YYYY-MM-DD)"
)
parser.add_argument(
"--breadth-50dma-date", default=None, help="Date of breadth 50DMA data (YYYY-MM-DD)"
)
parser.add_argument(
"--put-call-date", default=None, help="Date of put/call ratio data (YYYY-MM-DD)"
)
parser.add_argument(
"--margin-debt-date", default=None, help="Date of margin debt data (YYYY-MM-DD)"
)
# Additional context (not scored, but included in report)
parser.add_argument(
"--context",
nargs="*",
default=[],
help="Additional context items in 'key=value' format (e.g., 'Consumer Confidence=57.3')",
)
# Breadth auto-fetch control
parser.add_argument(
"--no-auto-breadth",
action="store_true",
help="Disable auto-fetch of 200DMA breadth from TraderMonty CSV",
)
# Leading stock basket mode
parser.add_argument(
"--static-basket",
action="store_true",
help="Use static default ETF basket instead of dynamic selection",
)
# Output
parser.add_argument("--output-dir", default="reports/", help="Output directory for reports")
parser.add_argument(
"--as-of",
type=_iso_date,
default=None,
help=(
"Evaluation date (YYYY-MM-DD). Historical live replay is rejected because "
"the quote provider is not point-in-time."
),
)
return parser.parse_args(argv)
def compute_data_freshness(date_args: dict, as_of_date: Optional[date] = None) -> dict:
"""
Compute data freshness factors for CLI input dates.
Uses business days instead of calendar days so that Friday data
remains fresh on Monday. Future dates are treated as anomalous input.
Args:
date_args: Dict with keys like 'breadth_200dma_date', 'breadth_50dma_date',
'put_call_date', 'margin_debt_date' -> YYYY-MM-DD strings.
Returns:
Dict with per-input freshness info and overall_confidence (min of all factors).
"""
from datetime import date as dateclass
from utils import count_business_days
freshness_map = {
"breadth_200dma_date": "breadth_200dma",
"breadth_50dma_date": "breadth_50dma",
"put_call_date": "put_call",
"margin_debt_date": "margin_debt",
}
result = {}
factors = []
today = as_of_date or dateclass.today()
for arg_key, label in freshness_map.items():
date_str = date_args.get(arg_key)
if not date_str:
continue
try:
d = dateclass.fromisoformat(date_str)
except (ValueError, TypeError):
result[label] = {"date": date_str, "age_days": None, "factor": 0.70}
factors.append(0.70)
continue
# Reject future dates as input anomaly
if d > today:
result[label] = {"date": date_str, "age_days": None, "factor": 0.70}
factors.append(0.70)
continue
biz_days = count_business_days(d, today)
if biz_days <= 1:
factor = 1.0
elif biz_days <= 3:
factor = 0.95
elif biz_days <= 7:
factor = 0.85
else:
factor = 0.70
result[label] = {"date": date_str, "age_days": biz_days, "factor": factor}
factors.append(factor)
result["overall_confidence"] = min(factors) if factors else 1.0
return result
def _load_previous_report(output_dir: str) -> Optional[dict]:
"""
Load the most recent market_top_*.json from output_dir.
Files are named market_top_YYYY-MM-DD_HHMMSS.json, so lexicographic
sorting gives chronological order.
"""
pattern = os.path.join(output_dir, "market_top_*.json")
files = sorted(glob.glob(pattern))
if not files:
return None
try:
with open(files[-1]) as f:
return json.load(f)
except (json.JSONDecodeError, OSError):
return None
def _compute_deltas(current_scores: dict[str, float], previous_report: Optional[dict]) -> dict:
"""
Compute delta between current and previous component scores.
Returns:
Dict with per-component delta info and composite delta.
If no previous report, direction is 'first_run' for all.
"""
component_keys = [
"distribution_days",
"leading_stocks",
"defensive_rotation",
"breadth_divergence",
"index_technical",
"sentiment",
]
deltas = {}
if previous_report is None:
for key in component_keys:
deltas[key] = {"delta": 0, "direction": "first_run"}
return {
"components": deltas,
"composite_delta": 0,
"composite_direction": "first_run",
"previous_date": None,
}
prev_components = previous_report.get("components", {})
prev_composite = previous_report.get("composite", {}).get("composite_score", 0)
for key in component_keys:
prev_score = prev_components.get(key, {}).get("score", 0)
curr_score = current_scores.get(key, 0)
delta = curr_score - prev_score
if abs(delta) <= 3:
direction = "stable"
elif delta > 0:
direction = "worsening"
else:
direction = "improving"
deltas[key] = {"delta": round(delta, 1), "direction": direction, "previous": prev_score}
prev_date = previous_report.get("metadata", {}).get("generated_at", None)
return {
"components": deltas,
"composite_delta": 0, # Will be filled after composite calc
"composite_direction": "first_run",
"previous_date": prev_date,
"previous_composite": prev_composite,
}
def _history_on_or_before(rows: list[dict], as_of_date: date) -> list[dict]:
"""Reject future bars at the consumer boundary; preserve provider order."""
filtered = []
for row in rows:
try:
row_date = date.fromisoformat(str(row.get("date", ""))[:10])
except (TypeError, ValueError):
continue
if row_date <= as_of_date:
filtered.append(row)
return filtered
def _batch_histories_on_or_before(
histories: dict[str, list[dict]], as_of_date: date
) -> dict[str, list[dict]]:
"""Apply the PIT ceiling to every symbol and discard empty histories."""
return {
symbol: filtered
for symbol, rows in histories.items()
if (filtered := _history_on_or_before(rows, as_of_date))
}
def _future_dated_scored_inputs(args: argparse.Namespace, as_of_date: date) -> list[str]:
"""Return scored CLI fields whose declared source date exceeds as-of."""
pairs = (
("breadth_200dma", "breadth_200dma_date"),
("breadth_50dma", "breadth_50dma_date"),
("put_call", "put_call_date"),
("margin_debt_yoy", "margin_debt_date"),
)
future = []
for value_name, date_name in pairs:
if getattr(args, value_name, None) is None:
continue
raw_date = getattr(args, date_name, None)
if not raw_date:
continue
try:
source_date = date.fromisoformat(raw_date)
except (TypeError, ValueError):
continue
if source_date > as_of_date:
future.append(value_name)
return future
def main(argv: Optional[list[str]] = None):
args = parse_arguments(argv)
live_today = datetime.now(ZoneInfo("America/New_York")).date()
as_of_date = args.as_of or live_today
if args.as_of is not None and as_of_date != live_today:
print(
"ERROR: historical --as-of requires point-in-time quote data, which the live "
"provider does not supply; refusing a look-ahead-biased replay",
file=sys.stderr,
)
return 2
future_inputs = _future_dated_scored_inputs(args, as_of_date)
if future_inputs:
print(
"ERROR: refusing future-dated scored input(s): " + ", ".join(future_inputs),
file=sys.stderr,
)
return 2
print("=" * 70)
print("Market Top Detector")
print("O'Neil (Distribution) + Minervini (Leadership) + Monty (Rotation)")
print("=" * 70)
print()
# Initialize FMP client
try:
client = FMPClient(api_key=args.api_key)
print("FMP API client initialized")
except ValueError as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
# ========================================================================
# Step 1: Fetch shared data (indices, ETFs)
# ========================================================================
print()
print("Step 1: Fetching Market Data")
print("-" * 70)
# S&P 500 data
print(" Fetching S&P 500 data...", end=" ", flush=True)
sp500_quote_list = client.get_quote("^GSPC")
sp500_quote = sp500_quote_list[0] if sp500_quote_list else None
sp500_history_data = client.get_historical_prices("^GSPC", days=260)
sp500_history = _history_on_or_before(
sp500_history_data.get("historical", []) if sp500_history_data else [],
as_of_date,
)
if sp500_quote and sp500_history:
print(f"OK (${sp500_quote.get('price', 0):.2f}, {len(sp500_history)} days)")
else:
print("FAILED")
print("ERROR: Cannot proceed without S&P 500 data", file=sys.stderr)
sys.exit(1)
# NASDAQ/QQQ data
print(" Fetching NASDAQ (QQQ) data...", end=" ", flush=True)
qqq_quote_list = client.get_quote("QQQ")
qqq_quote = qqq_quote_list[0] if qqq_quote_list else None
qqq_history_data = client.get_historical_prices("QQQ", days=260)
qqq_history = _history_on_or_before(
qqq_history_data.get("historical", []) if qqq_history_data else [],
as_of_date,
)
if qqq_quote and qqq_history:
print(f"OK (${qqq_quote.get('price', 0):.2f}, {len(qqq_history)} days)")
else:
print("WARN - NASDAQ data unavailable, using S&P 500 only")
# VIX
print(" Fetching VIX...", end=" ", flush=True)
vix_quote_list = client.get_quote("^VIX")
vix_quote = vix_quote_list[0] if vix_quote_list else None
vix_level = vix_quote.get("price", None) if vix_quote else None
if vix_level:
print(f"OK ({vix_level:.2f})")
else:
print("WARN - VIX unavailable")
# VIX Term Structure auto-detection
effective_vix_term = args.vix_term # CLI override takes priority
vix_term_auto = None
if effective_vix_term is None:
print(" Auto-detecting VIX term structure...", end=" ", flush=True)
vix_term_auto = client.get_vix_term_structure()
if vix_term_auto:
effective_vix_term = vix_term_auto["classification"]
print(f"OK ({effective_vix_term}, ratio={vix_term_auto['ratio']})")
else:
print("WARN - VIX3M unavailable, manual --vix-term needed")
# Leading ETFs (dynamic or static basket)
if args.static_basket:
selected_basket = list(LEADING_ETFS)
print(" Fetching Leading ETFs (static basket)...", end=" ", flush=True)
leading_quotes = client.get_batch_quotes(selected_basket)
leading_historical = client.get_batch_historical(selected_basket, days=60)
else:
print(" Fetching candidate pool quotes for dynamic basket...", end=" ", flush=True)
candidate_quotes = client.get_batch_quotes(CANDIDATE_POOL)
print(f"OK ({len(candidate_quotes)} candidates)")
selected_basket = select_dynamic_basket(candidate_quotes)
print(f" Selected dynamic basket: {selected_basket}")
print(" Fetching Leading ETFs (dynamic basket)...", end=" ", flush=True)
leading_quotes = {s: candidate_quotes[s] for s in selected_basket if s in candidate_quotes}
leading_historical = client.get_batch_historical(selected_basket, days=60)
leading_historical = _batch_histories_on_or_before(leading_historical, as_of_date)
print(f"OK ({len(leading_quotes)} quotes, {len(leading_historical)} histories)")
# Sector ETFs
all_sector_etfs = list(set(DEFENSIVE_ETFS + OFFENSIVE_ETFS))
# Remove QQQ from sector fetching if already fetched
sector_etfs_to_fetch = [e for e in all_sector_etfs if e != "QQQ"]
print(" Fetching Sector ETFs...", end=" ", flush=True)
sector_historical = _batch_histories_on_or_before(
client.get_batch_historical(sector_etfs_to_fetch, days=50),
as_of_date,
)
# Add QQQ history if available
if qqq_history:
sector_historical["QQQ"] = qqq_history[:50]
print(f"OK ({len(sector_historical)} ETFs)")
print()
# ========================================================================
# Step 2: Calculate Components
# ========================================================================
print("Step 2: Calculating Components")
print("-" * 70)
# Component 1: Distribution Days (25%)
print(" [1/6] Distribution Day Count...", end=" ", flush=True)
comp1 = calculate_distribution_days(sp500_history, qqq_history)
print(f"Score: {comp1['score']} ({comp1['signal']})")
# Component 2: Leading Stock Health (20%)
print(" [2/6] Leading Stock Health...", end=" ", flush=True)
comp2 = calculate_leading_stock_health(
leading_quotes, leading_historical, etf_list=selected_basket
)
print(f"Score: {comp2['score']} ({comp2['signal']})")
# Component 3: Defensive Rotation (15%)
print(" [3/6] Defensive Sector Rotation...", end=" ", flush=True)
comp3 = calculate_defensive_rotation(sector_historical)
print(f"Score: {comp3['score']} ({comp3['signal']})")
# Auto-fetch 200DMA breadth if not provided via CLI
effective_breadth_200dma = args.breadth_200dma
breadth_source = "cli"
breadth_auto_date = None
if effective_breadth_200dma is None and not args.no_auto_breadth:
print(" Fetching 200DMA breadth from TraderMonty CSV...", end=" ", flush=True)
auto_result = fetch_breadth_200dma(as_of_date=as_of_date)
if auto_result is not None:
effective_breadth_200dma = auto_result["value"]
breadth_source = "auto"
breadth_auto_date = auto_result["date"]
fresh_str = (
"fresh" if auto_result["is_fresh"] else f"STALE ({auto_result['days_old']}d old)"
)
print(f"OK ({effective_breadth_200dma}%, {auto_result['date']}, {fresh_str})")
if not auto_result["is_fresh"]:
print(f" WARNING: Breadth data is {auto_result['days_old']} days old")
else:
print("FAILED (will use neutral default)")
# Component 4: Breadth Divergence (15%)
print(" [4/6] Market Breadth Divergence...", end=" ", flush=True)
# Calculate index distance from 52-week high
sp500_year_high = sp500_quote.get("yearHigh", 0)
sp500_price = sp500_quote.get("price", 0)
if sp500_year_high > 0:
index_dist = (sp500_price - sp500_year_high) / sp500_year_high * 100
else:
index_dist = 0
comp4 = calculate_breadth_divergence(
breadth_200dma=effective_breadth_200dma,
breadth_50dma=args.breadth_50dma,
index_distance_from_high_pct=index_dist,
)
comp4["breadth_source"] = breadth_source
if breadth_auto_date:
comp4["breadth_auto_date"] = breadth_auto_date
print(f"Score: {comp4['score']} ({comp4['signal']})")
# Component 5: Index Technical (15%)
print(" [5/6] Index Technical Condition...", end=" ", flush=True)
comp5 = calculate_index_technical(
sp500_history, qqq_history, sp500_quote=sp500_quote, nasdaq_quote=qqq_quote
)
print(f"Score: {comp5['score']} ({comp5['signal']})")
# Component 6: Sentiment (10%)
print(" [6/6] Sentiment & Speculation...", end=" ", flush=True)
comp6 = calculate_sentiment(
vix_level=vix_level,
put_call_ratio=args.put_call,
vix_term_structure=effective_vix_term,
margin_debt_yoy_pct=args.margin_debt_yoy,
)
print(f"Score: {comp6['score']} ({comp6['signal']})")
print()
# Compute data freshness
freshness_args = {
"breadth_200dma_date": breadth_auto_date
if breadth_source == "auto"
else args.breadth_200dma_date,
"breadth_50dma_date": args.breadth_50dma_date,
"put_call_date": args.put_call_date,
"margin_debt_date": args.margin_debt_date,
}
data_freshness = compute_data_freshness(freshness_args, as_of_date=as_of_date)
# ========================================================================
# Step 3: Composite Score
# ========================================================================
print("Step 3: Calculating Composite Score")
print("-" * 70)
component_scores = {
"distribution_days": comp1["score"],
"leading_stocks": comp2["score"],
"defensive_rotation": comp3["score"],
"breadth_divergence": comp4["score"],
"index_technical": comp5["score"],
"sentiment": comp6["score"],
}
data_availability = {
"distribution_days": True, # Always available (requires S&P data to run)
"leading_stocks": comp2.get("data_available", True),
"defensive_rotation": comp3.get("data_available", True),
"breadth_divergence": comp4.get("data_available", True),
"index_technical": comp5.get("data_available", True),
"sentiment": comp6.get("data_available", True),
}
composite = calculate_composite_score(component_scores, data_availability)
print(f" Heuristic Risk Score: {composite['composite_score']}/100")
print(" Risk score, not an estimated event probability; budgets are policy heuristics.")
print(f" Risk Zone: {composite['zone']}")
print(f" Risk Budget: {composite['risk_budget']}")
print(
f" Strongest Warning: {composite['strongest_warning']['label']} "
f"({composite['strongest_warning']['score']})"
)
# Delta tracking vs previous run
previous_report = _load_previous_report(args.output_dir)
delta_info = _compute_deltas(component_scores, previous_report)
# Update composite delta now that we have the composite score
if previous_report is not None:
prev_composite = delta_info.get("previous_composite", 0)
comp_delta = composite["composite_score"] - prev_composite
delta_info["composite_delta"] = round(comp_delta, 1)
if abs(comp_delta) <= 3:
delta_info["composite_direction"] = "stable"
elif comp_delta > 0:
delta_info["composite_direction"] = "worsening"
else:
delta_info["composite_direction"] = "improving"
print(
f" vs Previous: {prev_composite} -> {composite['composite_score']} "
f"({delta_info['composite_delta']:+.1f})"
)
else:
print(" (First run - no comparison available)")
print()
# ========================================================================
# Step 4: Follow-Through Day Check
# ========================================================================
ftd = detect_follow_through_day(sp500_history, composite["composite_score"])
if ftd.get("applicable"):
print("Step 4: Follow-Through Day Monitor")
print("-" * 70)
print(f" {ftd['reason']}")
print()
# ========================================================================
# Step 5: Historical Comparison & Scenarios
# ========================================================================
print("Step 5: Historical Comparison & Scenarios")
print("-" * 70)
historical_comparison = compare_to_historical(component_scores, data_availability)
print(f" Historical comparison: {historical_comparison['narrative']}")
scenarios = generate_scenarios(component_scores, data_availability)
print(f" Generated {len(scenarios)} what-if scenarios")
print()
# ========================================================================
# Step 6: Generate Reports
# ========================================================================
print("Step 6: Generating Reports")
print("-" * 70)
# Parse additional context
additional_context = {}
for item in args.context:
if "=" in item:
key, value = item.split("=", 1)
additional_context[key.strip()] = value.strip()
# Build full analysis
analysis = {
"metadata": {
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"as_of_date": as_of_date.isoformat(),
"data_mode": "FMP API + CLI inputs",
"api_calls": client.get_api_stats(),
"cli_inputs": {
"breadth_200dma": effective_breadth_200dma,
"breadth_200dma_source": breadth_source,
"breadth_200dma_auto_date": breadth_auto_date,
"breadth_50dma": args.breadth_50dma,
"put_call_ratio": args.put_call,
"vix_term_structure": args.vix_term,
"margin_debt_yoy_pct": args.margin_debt_yoy,
},
"vix_term_auto": vix_term_auto,
"index_data": {
"sp500_price": sp500_price,
"sp500_year_high": sp500_year_high,
"sp500_distance_from_high_pct": round(index_dist, 2),
"qqq_price": qqq_quote.get("price", 0) if qqq_quote else None,
"vix_level": vix_level,
},
"data_freshness": data_freshness,
},
"composite": composite,
"components": {
"distribution_days": comp1,
"leading_stocks": comp2,
"defensive_rotation": comp3,
"breadth_divergence": comp4,
"index_technical": comp5,
"sentiment": comp6,
},
"follow_through_day": ftd,
"historical_comparison": historical_comparison,
"scenarios": scenarios,
"delta": delta_info,
"additional_context": additional_context,
}
timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
json_file = os.path.join(args.output_dir, f"market_top_{timestamp}.json")
md_file = os.path.join(args.output_dir, f"market_top_{timestamp}.md")
generate_json_report(analysis, json_file)
generate_markdown_report(analysis, md_file)
print()
print("=" * 70)
print("Market Top Detection Complete")
print("=" * 70)
print(f" Heuristic Risk Score: {composite['composite_score']}/100")
print(" Risk score, not an estimated event probability; budgets are policy heuristics.")
print(f" Risk Zone: {composite['zone']}")
print(f" Risk Budget: {composite['risk_budget']}")
print(f" JSON Report: {json_file}")
print(f" Markdown Report: {md_file}")
print()
stats = client.get_api_stats()
print("API Usage:")
print(f" API calls made: {stats['api_calls_made']}")
print(f" Cache entries: {stats['cache_entries']}")
print()
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/report_generator.py
#!/usr/bin/env python3
"""
Market Top Detector - Report Generator
Generates JSON and Markdown reports for market top detection analysis.
"""
import json
from typing import Optional
def generate_json_report(analysis: dict, output_file: str):
"""Save full analysis as JSON"""
with open(output_file, "w", encoding="utf-8") as f:
json.dump(analysis, f, indent=2, default=str)
print(f"JSON report saved to: {output_file}")
def _delta_arrow(delta_info: Optional[dict]) -> str:
"""Convert delta info to direction arrow."""
if delta_info is None:
return "---"
direction = delta_info.get("direction", "first_run")
delta = delta_info.get("delta", 0)
if direction == "first_run":
return "---"
elif direction == "worsening":
return f"+{delta:.0f} ↑"
elif direction == "improving":
return f"{delta:.0f} ↓"
else:
return "→"
def generate_markdown_report(analysis: dict, output_file: str):
"""Generate comprehensive Markdown report"""
lines = []
composite = analysis.get("composite", {})
components = analysis.get("components", {})
metadata = analysis.get("metadata", {})
ftd = analysis.get("follow_through_day", {})
delta = analysis.get("delta", {})
score = composite.get("composite_score", 0)
zone = composite.get("zone", "Unknown")
risk_budget = composite.get("risk_budget", "N/A")
# Header
lines.append("# Market Top Detector Report")
lines.append("")
lines.append(f"**Generated:** {metadata.get('generated_at', 'N/A')}")
lines.append(f"**Data Mode:** {metadata.get('data_mode', 'N/A')}")
lines.append("")
# Overall Assessment Box
lines.append("---")
lines.append("")
lines.append("## Overall Assessment")
lines.append("")
zone_emoji = _zone_emoji(composite.get("zone_color", ""))
lines.append("| Metric | Value |")
lines.append("|--------|-------|")
lines.append(f"| **Heuristic Risk Score** | **{score}/100** |")
lines.append(f"| **Risk Zone** | {zone_emoji} {zone} |")
lines.append(f"| **Risk Budget** | {risk_budget} |")
strongest = composite.get("strongest_warning", {})
weakest = composite.get("weakest_warning", {})
strongest_label = strongest.get("label", "N/A") or "N/A"
weakest_label = weakest.get("label", "N/A") or "N/A"
strongest_score = strongest.get("score", 0) or 0
weakest_score = weakest.get("score", 0) or 0
lines.append(f"| **Strongest Warning** | {strongest_label} ({strongest_score}/100) |")
lines.append(f"| **Weakest Warning** | {weakest_label} ({weakest_score}/100) |")
dq = composite.get("data_quality", {})
if dq:
lines.append(f"| **Data Quality** | {dq.get('label', 'N/A')} |")
lines.append("")
lines.append(
"> Risk score, not an estimated event probability. Risk budgets are policy heuristics."
)
lines.append("")
# Guidance
lines.append(f"> **Guidance:** {composite.get('guidance', '')}")
lines.append("")
# Component Scores Table
lines.append("---")
lines.append("")
lines.append("## Component Scores")
lines.append("")
lines.append(
"> **Reading Guide:** Higher score = higher market top risk. "
"A low score (e.g., Leading Stocks 20/100) means that component "
"is **healthy** and not signaling danger."
)
lines.append("")
delta_components = delta.get("components", {}) if delta else {}
has_delta = delta and delta.get("composite_direction") != "first_run"
if has_delta:
lines.append("| # | Component | Weight | Score | Δ | Contribution | Signal |")
lines.append("|---|-----------|--------|-------|---|--------------|--------|")
else:
lines.append("| # | Component | Weight | Score | Contribution | Signal |")
lines.append("|---|-----------|--------|-------|--------------|--------|")
component_order = [
"distribution_days",
"leading_stocks",
"defensive_rotation",
"breadth_divergence",
"index_technical",
"sentiment",
]
for i, key in enumerate(component_order, 1):
comp = composite.get("component_scores", {}).get(key, {})
detail = components.get(key, {})
signal = detail.get("signal", "N/A")
score_val = comp.get("score", 0)
weight_pct = f"{comp.get('weight', 0) * 100:.0f}%"
contribution = comp.get("weighted_contribution", 0)
bar = _score_bar(score_val)
if has_delta:
d_info = delta_components.get(key)
arrow = _delta_arrow(d_info)
lines.append(
f"| {i} | **{comp.get('label', key)}** | {weight_pct} | "
f"{bar} {score_val} | {arrow} | {contribution:.1f} | {signal} |"
)
else:
lines.append(
f"| {i} | **{comp.get('label', key)}** | {weight_pct} | "
f"{bar} {score_val} | {contribution:.1f} | {signal} |"
)
lines.append("")
# Delta summary
if has_delta:
prev_date = delta.get("previous_date", "N/A")
prev_composite = delta.get("previous_composite", 0)
comp_delta = delta.get("composite_delta", 0)
comp_dir = delta.get("composite_direction", "stable")
dir_arrow = "↑" if comp_dir == "worsening" else "↓" if comp_dir == "improving" else "→"
lines.append(
f"**vs. Previous Run ({prev_date}):** Score {prev_composite} → "
f"{score} ({comp_delta:+.1f} {dir_arrow})"
)
lines.append("")
elif delta and delta.get("composite_direction") == "first_run":
lines.append("*First run - no comparison available.*")
lines.append("")
# Detailed Component Breakdowns
lines.append("---")
lines.append("")
lines.append("## Component Details")
lines.append("")
# Component 1: Distribution Days
dist = components.get("distribution_days", {})
lines.append("### 1. Distribution Day Count")
lines.append("")
if dist:
sp = dist.get("sp500", {})
nq = dist.get("nasdaq", {})
lines.append(
f"- **Effective Count:** {dist.get('effective_count', 0)} "
f"(Primary: {dist.get('primary_index', 'N/A')})"
)
lines.append(
f"- **S&P 500:** {sp.get('distribution_days', 0)} distribution + "
f"{sp.get('stalling_days', 0)} stalling"
)
lines.append(
f"- **NASDAQ:** {nq.get('distribution_days', 0)} distribution + "
f"{nq.get('stalling_days', 0)} stalling"
)
# Distribution day details
for idx_name, idx_data in [("S&P 500", sp), ("NASDAQ", nq)]:
details = idx_data.get("details", [])
if details:
lines.append(f"\n**{idx_name} Distribution Events:**")
for d in details[:5]:
lines.append(
f" - {d.get('date', '?')}: {d.get('type', '?')} "
f"({d.get('pct_change', 0):+.2f}%, vol {d.get('volume_change', 0):+.1f}%)"
)
lines.append("")
# Component 2: Leading Stock Health
lead = components.get("leading_stocks", {})
lines.append("### 2. Leading Stock Health")
lines.append("")
if lead:
lines.append(f"- **ETFs Evaluated:** {lead.get('etfs_evaluated', 0)}")
lines.append(
f"- **ETFs Deteriorating:** {lead.get('etfs_deteriorating', 0)} "
f"({lead.get('deteriorating_pct', 0):.0f}%)"
)
lines.append(
f"- **Amplification Applied:** {'Yes (60%+ deteriorating)' if lead.get('amplified') else 'No'}"
)
partial_count = lead.get("partial_data_count", 0)
if partial_count > 0:
lines.append(
f"- **Note:** {partial_count} ETF(s) had insufficient history for full analysis."
)
etf_details = lead.get("etf_details", {})
if etf_details:
lines.append("")
lines.append("| ETF | Score | Distance from High | Flags |")
lines.append("|-----|-------|--------------------|-------|")
for sym, det in sorted(
etf_details.items(), key=lambda x: x[1].get("deterioration_score", 0), reverse=True
):
flags = det.get("flags", [])
flags_joined = "; ".join(flags)
flags_str = flags_joined if len(flags_joined) <= 80 else flags_joined[:77] + "..."
lines.append(
f"| {sym} | {det.get('deterioration_score', 0)} | "
f"{det.get('distance_from_high_pct', 0):+.1f}% | {flags_str} |"
)
lines.append("")
# Component 3: Defensive Rotation
defr = components.get("defensive_rotation", {})
lines.append("### 3. Defensive Sector Rotation")
lines.append("")
if defr:
lines.append(
f"- **Relative Performance (Def - Off):** "
f"{defr.get('relative_performance', 0):+.2f}% "
f"(over {defr.get('lookback_days', 20)} days)"
)
lines.append(f"- **Defensive Avg Return:** {defr.get('defensive_avg_return', 0):+.2f}%")
lines.append(f"- **Offensive Avg Return:** {defr.get('offensive_avg_return', 0):+.2f}%")
def_det = defr.get("defensive_details", {})
off_det = defr.get("offensive_details", {})
if def_det or off_det:
lines.append("")
lines.append("| Defensive ETF | Return | | Offensive ETF | Return |")
lines.append("|---------------|--------|-|---------------|--------|")
def_items = list(def_det.items())
off_items = list(off_det.items())
max_len = max(len(def_items), len(off_items))
for j in range(max_len):
d_sym = def_items[j][0] if j < len(def_items) else ""
d_ret = f"{def_items[j][1]:+.2f}%" if j < len(def_items) else ""
o_sym = off_items[j][0] if j < len(off_items) else ""
o_ret = f"{off_items[j][1]:+.2f}%" if j < len(off_items) else ""
lines.append(f"| {d_sym} | {d_ret} | | {o_sym} | {o_ret} |")
lines.append("")
# Component 4: Breadth
brd = components.get("breadth_divergence", {})
lines.append("### 4. Market Breadth Divergence")
lines.append("")
if brd:
breadth_source = brd.get("breadth_source", "cli")
if breadth_source == "auto":
auto_date = brd.get("breadth_auto_date", "N/A")
source_label = f"auto (TraderMonty CSV, {auto_date})"
else:
source_label = "manual (CLI input)"
lines.append(f"- **200DMA Breadth:** {brd.get('breadth_200dma', 'N/A')}% — {source_label}")
lines.append(f"- **50DMA Breadth:** {brd.get('breadth_50dma', 'N/A')}%")
lines.append(
f"- **Index Near Highs:** {'Yes' if brd.get('index_near_highs') else 'No'} "
f"({brd.get('index_distance_from_high_pct', 0):+.1f}% from 52wk high)"
)
lines.append(
f"- **Divergence Detected:** {'YES' if brd.get('divergence_detected') else 'No'}"
)
lines.append("")
# Component 5: Index Technical
tech = components.get("index_technical", {})
lines.append("### 5. Index Technical Condition")
lines.append("")
if tech:
for idx_name in ["sp500", "nasdaq"]:
idx = tech.get(idx_name, {})
if idx:
label = "S&P 500" if idx_name == "sp500" else "NASDAQ"
lines.append(f"**{label}:** (Score: {idx.get('raw_score', 0)})")
for flag in idx.get("flags", []):
lines.append(f" - {flag}")
mas = idx.get("mas", {})
if mas:
ma_str = ", ".join([f"{k}: ${v}" for k, v in mas.items()])
lines.append(f" - Moving Averages: {ma_str}")
lines.append("")
lines.append("")
# Component 6: Sentiment
sent = components.get("sentiment", {})
lines.append("### 6. Sentiment & Speculation")
lines.append("")
if sent:
details = sent.get("details", {})
pc = details.get("put_call_ratio", {})
vix = details.get("vix_level", {})
vts = details.get("vix_term_structure", {})
md = details.get("margin_debt", {})
lines.append(
f"- **Put/Call Ratio:** {pc.get('value', 'N/A')} "
f"(+{pc.get('points', 0)}pt) - {pc.get('interpretation', '')}"
)
lines.append(
f"- **VIX Level:** {vix.get('value', 'N/A')} "
f"({vix.get('points', 0):+d}pt) - {vix.get('interpretation', '')}"
)
lines.append(
f"- **VIX Term Structure:** {vts.get('value', 'N/A')} "
f"({vts.get('points', 0):+d}pt) - {vts.get('interpretation', '')}"
)
if md:
lines.append(
f"- **Margin Debt YoY:** {md.get('yoy_pct', 'N/A')}% "
f"({md.get('points', 0):+d}pt) - {md.get('interpretation', '')}"
)
lines.append("")
# Follow-Through Day Monitor
if ftd and ftd.get("applicable"):
lines.append("---")
lines.append("")
lines.append("## Follow-Through Day Monitor")
lines.append("")
if ftd.get("ftd_detected"):
lines.append(f"**FTD DETECTED** on {ftd.get('ftd_day', 'N/A')}")
else:
lines.append(f"- Rally Day Count: {ftd.get('rally_day_count', 0)}")
lines.append(f"- {ftd.get('reason', '')}")
lines.append("")
# Recommended Actions
lines.append("---")
lines.append("")
lines.append("## Recommended Actions")
lines.append("")
lines.append(f"**Risk Zone:** {zone}")
lines.append(f"**Risk Budget:** {risk_budget}")
lines.append("")
for action in composite.get("actions", []):
lines.append(f"- {action}")
lines.append("")
# Historical Comparison
hist_comp = analysis.get("historical_comparison", {})
if hist_comp:
lines.append("---")
lines.append("")
lines.append("## Historical Comparison")
lines.append("")
lines.append(
f"**Closest Pattern:** {hist_comp.get('closest_match') or 'No comparable episode'} "
f"(SSD: {hist_comp.get('closest_ssd') if hist_comp.get('closest_ssd') is not None else 'N/A'})"
)
lines.append("")
lines.append(f"> {hist_comp.get('narrative', '')}")
lines.append("")
comparisons = hist_comp.get("comparisons", [])
if comparisons:
lines.append("| Historical Top | SSD (lower = closer) |")
lines.append("|---------------|---------------------|")
for c in comparisons:
lines.append(f"| {c['name']} | {c['ssd']} |")
lines.append("")
# What-If Scenarios
scenarios = analysis.get("scenarios", [])
if scenarios:
lines.append("---")
lines.append("")
lines.append("## What-If Scenarios")
lines.append("")
lines.append("| Scenario | Description | New Score | Zone | Delta |")
lines.append("|----------|-------------|-----------|------|-------|")
for s in scenarios:
delta_str = f"{s['delta']:+.1f}" if s["delta"] != 0 else "0.0"
lines.append(
f"| **{s['name']}** | {s['description']} | "
f"{s['new_score']} | {s['new_zone']} | {delta_str} |"
)
lines.append("")
# Additional Context
extra = analysis.get("additional_context", {})
if extra:
lines.append("---")
lines.append("")
lines.append("## Additional Context (Not Scored)")
lines.append("")
for key, value in extra.items():
lines.append(f"- **{key}:** {value}")
lines.append("")
# Methodology
lines.append("---")
lines.append("")
lines.append("## Methodology")
lines.append("")
lines.append("This analysis integrates three complementary market top detection approaches:")
lines.append("")
lines.append(
"1. **O'Neil (Distribution Days):** Institutional selling pressure via volume-confirmed declines"
)
lines.append(
"2. **Minervini (Leading Stock Deterioration):** Growth leadership breakdown pattern"
)
lines.append(
"3. **Monty (Defensive Rotation):** Capital flow from offensive to defensive sectors"
)
lines.append("")
lines.append(
"Additional components (Breadth, Technical, Sentiment) provide confirmation signals."
)
lines.append("Composite score is a weighted average of all 6 components (0-100 scale).")
lines.append("")
lines.append("For detailed methodology, see `references/market_top_methodology.md`.")
lines.append("")
# Disclaimer
lines.append("---")
lines.append("")
lines.append(
"**Disclaimer:** This analysis is for educational and informational purposes only. "
"Not investment advice. Past patterns may not predict future outcomes. "
"Conduct your own research and consult a financial advisor before making "
"investment decisions."
)
lines.append("")
with open(output_file, "w", encoding="utf-8") as f:
f.write("\n".join(lines))
print(f"Markdown report saved to: {output_file}")
def _zone_emoji(color: str) -> str:
mapping = {
"green": "🟢",
"yellow": "🟡",
"orange": "🟠",
"red": "🔴",
"critical": "⚫",
}
return mapping.get(color, "⚪")
def _score_bar(score: int) -> str:
"""Simple text bar for score visualization"""
if score >= 80:
return "████"
elif score >= 60:
return "███░"
elif score >= 40:
return "██░░"
elif score >= 20:
return "█░░░"
else:
return "░░░░"
scripts/scenario_engine.py
#!/usr/bin/env python3
"""
What-If Scenario Analysis Engine
Generates 4 sensitivity scenarios by perturbing component scores
and re-running the composite scoring engine.
"""
from scorer import calculate_composite_score
def generate_scenarios(
component_scores: dict[str, float], data_availability: dict[str, bool] = None
) -> list[dict]:
"""
Generate 4 what-if scenarios based on current component scores.
Scenarios:
1. Breadth Deterioration: breadth_divergence -> 55
2. Distribution Reset: distribution_days -> 55
3. Full Deterioration: top 3 weakest components each +20pt
4. Recovery: top 2 strongest components each -30pt
Args:
component_scores: Dict of component key -> score (0-100)
data_availability: Optional data availability dict
Returns:
List of 4 scenario dicts with name, changes, new_score, new_zone
"""
scenarios = []
# Scenario 1: Breadth Deterioration
s1 = dict(component_scores)
s1["breadth_divergence"] = max(s1.get("breadth_divergence", 0), 55)
s1_result = calculate_composite_score(s1, data_availability)
scenarios.append(
{
"name": "Breadth Deterioration",
"description": "Breadth divergence worsens to 55+",
"changes": {"breadth_divergence": s1["breadth_divergence"]},
"new_score": s1_result["composite_score"],
"new_zone": s1_result["zone"],
"delta": round(
s1_result["composite_score"] - _current_score(component_scores, data_availability),
1,
),
}
)
# Scenario 2: Distribution Reset (ease to 55 or less)
s2 = dict(component_scores)
s2["distribution_days"] = min(s2.get("distribution_days", 0), 55)
s2_result = calculate_composite_score(s2, data_availability)
scenarios.append(
{
"name": "Distribution Reset",
"description": "Distribution days ease to 55 or less",
"changes": {"distribution_days": s2["distribution_days"]},
"new_score": s2_result["composite_score"],
"new_zone": s2_result["zone"],
"delta": round(
s2_result["composite_score"] - _current_score(component_scores, data_availability),
1,
),
}
)
# Scenario 3: Full Deterioration - raise lowest 3 by +20pt
sorted_components = sorted(component_scores.items(), key=lambda x: x[1])
s3 = dict(component_scores)
changes_3 = {}
for key, val in sorted_components[:3]:
new_val = min(100, val + 20)
s3[key] = new_val
changes_3[key] = new_val
s3_result = calculate_composite_score(s3, data_availability)
scenarios.append(
{
"name": "Full Deterioration",
"description": "3 weakest components each worsen by +20pt",
"changes": changes_3,
"new_score": s3_result["composite_score"],
"new_zone": s3_result["zone"],
"delta": round(
s3_result["composite_score"] - _current_score(component_scores, data_availability),
1,
),
}
)
# Scenario 4: Recovery - lower highest 2 by -30pt
sorted_desc = sorted(component_scores.items(), key=lambda x: x[1], reverse=True)
s4 = dict(component_scores)
changes_4 = {}
for key, val in sorted_desc[:2]:
new_val = max(0, val - 30)
s4[key] = new_val
changes_4[key] = new_val
s4_result = calculate_composite_score(s4, data_availability)
scenarios.append(
{
"name": "Recovery",
"description": "2 strongest warning signals improve by -30pt",
"changes": changes_4,
"new_score": s4_result["composite_score"],
"new_zone": s4_result["zone"],
"delta": round(
s4_result["composite_score"] - _current_score(component_scores, data_availability),
1,
),
}
)
return scenarios
def _current_score(
component_scores: dict[str, float], data_availability: dict[str, bool] = None
) -> float:
"""Helper to get the current composite score."""
result = calculate_composite_score(component_scores, data_availability)
return result["composite_score"]
scripts/scorer.py
#!/usr/bin/env python3
"""
Market Top Detector - Composite Scoring Engine
Combines 6 component scores into a weighted composite (0-100).
Component Weights:
1. Distribution Day Count: 25%
2. Leading Stock Health: 20%
3. Defensive Sector Rotation: 15%
4. Market Breadth Divergence: 15%
5. Index Technical Condition: 15%
6. Sentiment & Speculation: 10%
Total: 100%
Risk Zone Mapping:
0-20: Green (Normal) - Risk Budget: 100%
21-40: Yellow (Early Warning) - Risk Budget: 80-90%
41-60: Orange (Elevated Risk) - Risk Budget: 60-75%
61-80: Red (High Risk)- Risk Budget: 40-55%
81-100:Critical(Very High Risk) - Risk Budget: 20-35%
"""
from typing import Optional
COMPONENT_WEIGHTS = {
"distribution_days": 0.25,
"leading_stocks": 0.20,
"defensive_rotation": 0.15,
"breadth_divergence": 0.15,
"index_technical": 0.15,
"sentiment": 0.10,
}
COMPONENT_LABELS = {
"distribution_days": "Distribution Day Count",
"leading_stocks": "Leading Stock Health",
"defensive_rotation": "Defensive Sector Rotation",
"breadth_divergence": "Market Breadth Divergence",
"index_technical": "Index Technical Condition",
"sentiment": "Sentiment & Speculation",
}
# Correlated component pairs: when both are extreme, discount the lower-weight one
CORRELATION_PAIRS = [("distribution_days", "defensive_rotation")]
CORRELATION_THRESHOLD = 80
CORRELATION_DISCOUNT = 0.8
def _apply_correlation_adjustment(component_scores: dict[str, float]) -> dict:
"""
Discount correlated components when both are extreme.
When both components in a CORRELATION_PAIR are >= CORRELATION_THRESHOLD,
the lower-weight component gets multiplied by CORRELATION_DISCOUNT.
Returns:
Dict with 'adjusted_scores' and 'adjustments' details.
"""
adjusted = dict(component_scores)
adjustments = []
for comp_a, comp_b in CORRELATION_PAIRS:
score_a = component_scores.get(comp_a, 0)
score_b = component_scores.get(comp_b, 0)
if score_a >= CORRELATION_THRESHOLD and score_b >= CORRELATION_THRESHOLD:
# Discount the lower-weight component
weight_a = COMPONENT_WEIGHTS.get(comp_a, 0)
weight_b = COMPONENT_WEIGHTS.get(comp_b, 0)
if weight_a >= weight_b:
discounted = comp_b
original = score_b
else:
discounted = comp_a
original = score_a
adjusted[discounted] = original * CORRELATION_DISCOUNT
adjustments.append(
{
"pair": [comp_a, comp_b],
"discounted_component": discounted,
"original_score": original,
"adjusted_score": adjusted[discounted],
"discount_factor": CORRELATION_DISCOUNT,
}
)
return {"adjusted_scores": adjusted, "adjustments": adjustments}
def calculate_composite_score(
component_scores: dict[str, float], data_availability: Optional[dict[str, bool]] = None
) -> dict:
"""
Calculate weighted composite heuristic market top risk score.
Args:
component_scores: Dict with keys matching COMPONENT_WEIGHTS,
each value 0-100
data_availability: Optional dict mapping component key -> bool indicating
if data was actually available (vs neutral default)
Returns:
Dict with composite_score, zone, risk_budget, guidance,
weakest/strongest components, component breakdown, and data_quality
"""
if data_availability is None:
data_availability = {}
# Apply correlation adjustment
corr_result = _apply_correlation_adjustment(component_scores)
adjusted_scores = corr_result["adjusted_scores"]
correlation_adjustment = corr_result["adjustments"]
# Build effective weights once: redistribute from unavailable to available
available_weight = 0.0
for key, weight in COMPONENT_WEIGHTS.items():
if data_availability.get(key, True):
available_weight += weight
effective_weights = {}
for key, weight in COMPONENT_WEIGHTS.items():
if data_availability.get(key, True):
effective_weights[key] = weight / available_weight if available_weight > 0 else weight
# Calculate weighted composite using adjusted scores and effective weights
composite = 0.0
for key in effective_weights:
score = adjusted_scores.get(key, 0)
composite += score * effective_weights[key]
composite = round(composite, 1)
# Identify strongest and weakest from available components only
available_scores = {k: v for k, v in component_scores.items() if k in effective_weights}
if available_scores:
strongest_warning = max(available_scores, key=available_scores.get)
weakest_warning = min(available_scores, key=available_scores.get)
else:
strongest_warning = None
weakest_warning = None
# Get zone interpretation
zone_info = _interpret_zone(composite)
# Calculate data quality
available_count = sum(
1
for k in COMPONENT_WEIGHTS
if data_availability.get(k, True) # Default True for backward compat
)
total_components = len(COMPONENT_WEIGHTS)
missing_components = [
COMPONENT_LABELS[k] for k in COMPONENT_WEIGHTS if not data_availability.get(k, True)
]
if available_count == total_components:
quality_label = f"Complete ({available_count}/{total_components} components)"
elif available_count >= total_components - 2:
quality_label = (
f"Partial ({available_count}/{total_components} components) - interpret with caution"
)
else:
quality_label = (
f"Limited ({available_count}/{total_components} components) - low confidence"
)
data_quality = {
"available_count": available_count,
"total_components": total_components,
"label": quality_label,
"missing_components": missing_components,
}
# Build strongest/weakest dicts (handle None when no available components)
if strongest_warning is not None:
strongest_dict = {
"component": strongest_warning,
"label": COMPONENT_LABELS.get(strongest_warning, strongest_warning),
"score": available_scores.get(strongest_warning, 0),
}
else:
strongest_dict = {"component": None, "label": "N/A", "score": 0}
if weakest_warning is not None:
weakest_dict = {
"component": weakest_warning,
"label": COMPONENT_LABELS.get(weakest_warning, weakest_warning),
"score": available_scores.get(weakest_warning, 0),
}
else:
weakest_dict = {"component": None, "label": "N/A", "score": 0}
return {
"score_kind": "heuristic_risk_score",
"score_version": "market-top-risk-v1",
"calibrated_probability": False,
"interpretation": "Risk score, not an estimated event probability. Risk budgets are policy heuristics.",
"composite_score": composite,
"zone": zone_info["zone"],
"zone_color": zone_info["color"],
"risk_budget": zone_info["risk_budget"],
"guidance": zone_info["guidance"],
"actions": zone_info["actions"],
"strongest_warning": strongest_dict,
"weakest_warning": weakest_dict,
"data_quality": data_quality,
"correlation_adjustment": correlation_adjustment,
"component_scores": {
k: {
"score": component_scores.get(k, 0),
"adjusted_score": adjusted_scores.get(k, component_scores.get(k, 0)),
"weight": w,
"weighted_contribution": round(adjusted_scores.get(k, 0) * effective_weights[k], 1)
if k in effective_weights
else 0.0,
"label": COMPONENT_LABELS[k],
}
for k, w in COMPONENT_WEIGHTS.items()
},
}
def _interpret_zone(composite: float) -> dict:
"""Map composite score to risk zone"""
if composite <= 20:
return {
"zone": "Green (Normal)",
"color": "green",
"risk_budget": "100%",
"guidance": "Normal market conditions. Maintain standard position management.",
"actions": [
"Normal position sizing",
"Standard stop-loss levels",
"New position entries allowed",
],
}
elif composite <= 40:
return {
"zone": "Yellow (Early Warning)",
"color": "yellow",
"risk_budget": "80-90%",
"guidance": "Early warning signs detected. Tighten stops and reduce new entries.",
"actions": [
"Tighten stop-losses by 10-20%",
"Reduce new position sizes by 25-50%",
"Review weakest positions for exits",
"Monitor distribution days closely",
],
}
elif composite <= 60:
return {
"zone": "Orange (Elevated Risk)",
"color": "orange",
"risk_budget": "60-75%",
"guidance": "Elevated risk of correction. Begin profit-taking on weaker positions.",
"actions": [
"Take profits on weakest 25-30% of positions",
"No new momentum entries",
"Only quality stocks near support",
"Raise cash allocation",
"Watch for Follow-Through Day if market pulls back",
],
}
elif composite <= 80:
return {
"zone": "Red (High Risk)",
"color": "red",
"risk_budget": "40-55%",
"guidance": "High heuristic market top risk. Aggressive profit-taking recommended.",
"actions": [
"Aggressive profit-taking (sell 40-50% of positions)",
"Maximum cash allocation",
"Only hold strongest leaders",
"Consider hedges (put options, inverse ETFs)",
"Prepare short watchlist",
],
}
else:
return {
"zone": "Critical (Very High Risk)",
"color": "critical",
"risk_budget": "20-35%",
"guidance": "Very high heuristic market top risk. Maximum defensive posture.",
"actions": [
"Sell most positions (keep only 20-35% invested)",
"Full hedge implementation",
"Short positions on weakest leaders",
"Preserve capital as primary objective",
"Watch for capitulation/Follow-Through Day for re-entry",
],
}
def detect_follow_through_day(index_history: list[dict], composite_score: float) -> dict:
"""
Detect Follow-Through Day (FTD) signal for bottom confirmation.
Only relevant when composite > 40 (Orange zone or worse).
O'Neil's Strict FTD Rules:
1. Identify swing low: significant decline (3+ down days, -3%+ from recent high)
2. Rally Day 1: first up day (close > previous close) after swing low
3. FTD: Day 4-7 of rally, gain >= 1.5% on volume higher than previous day
4. Rally resets if price closes below swing low
Args:
index_history: Daily OHLCV (most recent first)
composite_score: Current composite score
Returns:
Dict with ftd_detected, rally_day_count, details
"""
if composite_score < 40:
return {
"ftd_detected": False,
"applicable": False,
"reason": "Composite < 40 (Green/Yellow zone) - FTD monitoring not needed",
}
if not index_history or len(index_history) < 10:
return {
"ftd_detected": False,
"applicable": True,
"reason": "Insufficient data for FTD analysis",
}
# Work in chronological order (oldest first)
history = list(reversed(index_history))
n = len(history)
# Only look at the most recent 40 trading days
lookback = min(40, n)
history = history[n - lookback :]
n = len(history)
# Step 1: Find swing low within the lookback window
# Swing low = lowest close after a decline of 3%+ from a recent high
swing_low_idx = None
swing_low_price = None
# Scan for the most recent swing low
for i in range(n - 1, 2, -1): # from recent to old
low_close = history[i].get("close", 0)
if low_close <= 0:
continue
# Look back from this point for a recent high (within 20 days)
search_start = max(0, i - 20)
recent_high = 0
for j in range(search_start, i):
c = history[j].get("close", 0)
if c > recent_high:
recent_high = c
if recent_high <= 0:
continue
decline_pct = (low_close - recent_high) / recent_high * 100
# Check for 3%+ decline from recent high
if decline_pct <= -3.0:
# Verify at least 3 down days in the decline
down_days = 0
for j in range(search_start + 1, i + 1):
prev_c = history[j - 1].get("close", 0)
curr_c = history[j].get("close", 0)
if prev_c > 0 and curr_c < prev_c:
down_days += 1
if down_days >= 3:
# Check this is actually a local low
is_local_low = True
if i > 0 and history[i - 1].get("close", 0) < low_close:
is_local_low = False
if i + 1 < n and history[i + 1].get("close", 0) < low_close:
is_local_low = False
if is_local_low:
swing_low_idx = i
swing_low_price = low_close
break
if swing_low_idx is None:
return {
"ftd_detected": False,
"applicable": True,
"reason": "No qualifying swing low found (need 3%+ decline with 3+ down days)",
"rally_day_count": 0,
}
# Step 2 & 3: Find Rally Day 1 and count rally days with reset handling
# Uses seeking_rally_day_1 flag to properly re-seek Day 1 after resets.
# TODO: FTD logic here mirrors ftd-detector skill's state machine.
# Consider extracting shared FTD core if both need changes.
rally_day_1_idx = None
rally_day_count = 0
ftd_detected = False
ftd_day = None
seeking_rally_day_1 = True
reset_count = 0
for i in range(swing_low_idx + 1, n):
curr_close = history[i].get("close", 0)
prev_close = history[i - 1].get("close", 0) if i > 0 else 0
# Check for rally reset: price closes below swing low
if curr_close < swing_low_price:
swing_low_idx = i
swing_low_price = curr_close
rally_day_count = 0
ftd_detected = False
seeking_rally_day_1 = True
reset_count += 1
rally_day_1_idx = None
continue
# Seek Rally Day 1: first up close (or range top 50%) after swing low
if seeking_rally_day_1:
if prev_close <= 0:
continue
high = history[i].get("high", curr_close)
low = history[i].get("low", curr_close)
bar_range = high - low
range_position = (curr_close - low) / bar_range if bar_range > 0 else 0
if curr_close > prev_close or range_position >= 0.5:
rally_day_1_idx = i
rally_day_count = 1
seeking_rally_day_1 = False
continue
# Normal rally day counting (all days after Day 1 count)
rally_day_count += 1
# Check FTD on days 4-7
if 4 <= rally_day_count <= 7 and prev_close > 0:
gain_pct = (curr_close - prev_close) / prev_close * 100
curr_volume = history[i].get("volume", 0)
prev_volume = history[i - 1].get("volume", 0)
if gain_pct >= 1.5 and prev_volume > 0 and curr_volume > prev_volume:
ftd_detected = True
ftd_day = history[i].get("date", f"day-{i}")
break
if rally_day_1_idx is None:
return {
"ftd_detected": False,
"applicable": True,
"reason": "No rally attempt started after swing low",
"rally_day_count": 0,
"swing_low_date": history[swing_low_idx].get("date", "N/A"),
"swing_low_price": swing_low_price,
"reset_count": reset_count,
}
# Cap rally_day_count for reporting
days_since_rally_start = n - rally_day_1_idx
rally_day_count = min(rally_day_count, days_since_rally_start)
swing_low_date = history[swing_low_idx].get("date", "N/A")
if ftd_detected:
reason = f"Follow-Through Day detected on {ftd_day} (Day {rally_day_count} of rally from {swing_low_date} low)"
elif rally_day_count >= 7:
reason = (
f"Rally attempt: Day {rally_day_count} from {swing_low_date} low - "
"FTD window (Day 4-7) passed without qualifying day"
)
else:
reason = (
f"Rally attempt: Day {rally_day_count} from {swing_low_date} low "
f"(FTD requires Day 4-7 with >= 1.5% gain on higher volume)"
)
return {
"ftd_detected": ftd_detected,
"applicable": True,
"rally_day_count": rally_day_count,
"ftd_day": ftd_day,
"swing_low_date": swing_low_date,
"swing_low_price": swing_low_price,
"reset_count": reset_count,
"reason": reason,
}
# Testing
if __name__ == "__main__":
print("Testing Market Top Scorer...\n")
# Test 1: Moderate risk scenario (calibration target: ~50)
test_scores = {
"distribution_days": 45,
"leading_stocks": 52,
"defensive_rotation": 82,
"breadth_divergence": 20,
"index_technical": 42,
"sentiment": 62,
}
result = calculate_composite_score(test_scores)
print("Test 1 - Moderate Risk:")
print(f" Composite: {result['composite_score']}/100")
print(f" Zone: {result['zone']}")
print(f" Risk Budget: {result['risk_budget']}")
print(
f" Strongest Warning: {result['strongest_warning']['label']} "
f"({result['strongest_warning']['score']})"
)
print()
# Test 2: Healthy market
healthy = {
"distribution_days": 0,
"leading_stocks": 10,
"defensive_rotation": 0,
"breadth_divergence": 10,
"index_technical": 5,
"sentiment": 15,
}
result2 = calculate_composite_score(healthy)
print("Test 2 - Healthy Market:")
print(f" Composite: {result2['composite_score']}/100")
print(f" Zone: {result2['zone']}")
print()
# Test 3: Crisis
crisis = {
"distribution_days": 100,
"leading_stocks": 90,
"defensive_rotation": 85,
"breadth_divergence": 80,
"index_technical": 75,
"sentiment": 70,
}
result3 = calculate_composite_score(crisis)
print("Test 3 - Crisis:")
print(f" Composite: {result3['composite_score']}/100")
print(f" Zone: {result3['zone']}")
print()
# Test 4: Data quality tracking
partial_scores = {
"distribution_days": 45,
"leading_stocks": 52,
"defensive_rotation": 50,
"breadth_divergence": 50,
"index_technical": 42,
"sentiment": 50,
}
partial_availability = {
"distribution_days": True,
"leading_stocks": True,
"defensive_rotation": False,
"breadth_divergence": False,
"index_technical": True,
"sentiment": False,
}
result4 = calculate_composite_score(partial_scores, partial_availability)
print("Test 4 - Partial Data:")
print(f" Composite: {result4['composite_score']}/100")
print(f" Data Quality: {result4['data_quality']['label']}")
print(f" Missing: {result4['data_quality']['missing_components']}")
print()
print("All tests completed.")
scripts/tests/conftest.py
"""Shared fixtures for Market Top Detector tests"""
import os
import sys
# Add scripts directory to path so calculators can be imported
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
# Add tests directory to path so helpers can be imported
sys.path.insert(0, os.path.dirname(__file__))
scripts/tests/helpers.py
"""Importable test helpers for Market Top Detector tests"""
def make_daily_bar(close, volume=1000000, date="2026-01-15", open_=None, high=None, low=None):
"""Helper to create a daily OHLCV bar dict."""
if open_ is None:
open_ = close
if high is None:
high = close * 1.005
if low is None:
low = close * 0.995
return {
"date": date,
"open": open_,
"high": high,
"low": low,
"close": close,
"adjClose": close,
"volume": volume,
}
def make_history(closes, base_volume=1000000, start_date="2026-01-15"):
"""Create a list of daily bars from a list of closes (most recent first)."""
bars = []
for i, close in enumerate(closes):
vol = base_volume + (i * 1000) # Slight volume variation
bars.append(
make_daily_bar(
close=close,
volume=vol,
date=f"day-{i}",
)
)
return bars
def make_history_with_volumes(close_volume_pairs, start_date="2026-01-15"):
"""Create bars from list of (close, volume) tuples (most recent first)."""
bars = []
for i, (close, volume) in enumerate(close_volume_pairs):
bars.append(
make_daily_bar(
close=close,
volume=volume,
date=f"day-{i}",
)
)
return bars
scripts/tests/test_breadth_csv_client.py
"""Tests for Breadth CSV Client (TraderMonty auto-fetch)."""
from datetime import date, datetime
from unittest.mock import MagicMock, patch
import requests
from breadth_csv_client import fetch_breadth_200dma
# Sample CSV content matching TraderMonty's format
SAMPLE_CSV = (
"Date,S&P500_Price,Breadth_Index_Raw,Breadth_Index_200MA,"
"Breadth_Index_8MA,Breadth_200MA_Trend,Bearish_Signal,"
"Is_Peak,Is_Trough,Is_Trough_8MA_Below_04\n"
"2026-02-14,6100.50,0.6200,0.5800,0.6300,1,False,False,False,False\n"
"2026-02-18,6150.25,0.6481,0.5850,0.6400,1,False,False,False,False\n"
)
def _mock_response(text, status_code=200):
resp = MagicMock()
resp.text = text
resp.status_code = status_code
resp.raise_for_status = MagicMock()
return resp
class TestFetchBreadth200dma:
@patch("breadth_csv_client.requests.get")
def test_fetch_success_returns_value(self, mock_get):
"""Normal fetch returns a dict with value in 0-100 range."""
mock_get.return_value = _mock_response(SAMPLE_CSV)
result = fetch_breadth_200dma()
assert result is not None
assert 0 <= result["value"] <= 100
assert result["source"] == "TraderMonty CSV"
@patch("breadth_csv_client.requests.get")
def test_fetch_scales_raw_to_percent(self, mock_get):
"""Breadth_Index_Raw=0.6481 should become value=64.81."""
mock_get.return_value = _mock_response(SAMPLE_CSV)
result = fetch_breadth_200dma()
assert result["value"] == 64.81
@patch("breadth_csv_client.requests.get")
def test_fetch_returns_latest_date(self, mock_get):
"""Should return the date of the latest (last) row."""
mock_get.return_value = _mock_response(SAMPLE_CSV)
result = fetch_breadth_200dma()
assert result["date"] == "2026-02-18"
@patch("breadth_csv_client.requests.get")
def test_as_of_selects_latest_non_future_row(self, mock_get):
mock_get.return_value = _mock_response(SAMPLE_CSV)
result = fetch_breadth_200dma(as_of_date=date(2026, 2, 16))
assert result is not None
assert result["date"] == "2026-02-14"
assert result["value"] == 62.0
@patch("breadth_csv_client.requests.get")
def test_as_of_with_only_future_rows_returns_none(self, mock_get):
mock_get.return_value = _mock_response(SAMPLE_CSV)
result = fetch_breadth_200dma(as_of_date=date(2026, 2, 13))
assert result is None
@patch("breadth_csv_client.requests.get")
@patch("breadth_csv_client.datetime")
def test_fetch_freshness_check(self, mock_dt, mock_get):
"""Data from today should be is_fresh=True."""
mock_get.return_value = _mock_response(SAMPLE_CSV)
mock_dt.strptime = datetime.strptime
mock_dt.now.return_value = datetime(2026, 2, 18, 12, 0, 0)
result = fetch_breadth_200dma()
assert result["is_fresh"] is True
assert result["days_old"] == 0
@patch("breadth_csv_client.requests.get")
def test_fetch_failure_returns_none(self, mock_get):
"""Network error should return None."""
mock_get.side_effect = requests.ConnectionError("Connection refused")
result = fetch_breadth_200dma()
assert result is None
@patch("breadth_csv_client.requests.get")
def test_fetch_empty_csv_returns_none(self, mock_get):
"""CSV with only header (no data rows) should return None."""
header_only = (
"Date,S&P500_Price,Breadth_Index_Raw,Breadth_Index_200MA,"
"Breadth_Index_8MA,Breadth_200MA_Trend,Bearish_Signal,"
"Is_Peak,Is_Trough,Is_Trough_8MA_Below_04\n"
)
mock_get.return_value = _mock_response(header_only)
result = fetch_breadth_200dma()
assert result is None
class TestCheckDataFreshness:
"""Test _check_data_freshness function."""
def test_future_date_returns_not_fresh(self):
"""Future date should return is_fresh=False, days_old=None."""
from breadth_csv_client import _check_data_freshness
# Use a date far in the future
result = _check_data_freshness("2099-12-31")
assert result["is_fresh"] is False
assert result["days_old"] is None
def test_invalid_date_returns_not_fresh(self):
"""Invalid date string should return is_fresh=False, days_old=None."""
from breadth_csv_client import _check_data_freshness
result = _check_data_freshness("not-a-date")
assert result["is_fresh"] is False
assert result["days_old"] is None
scripts/tests/test_breadth.py
"""Tests for Breadth Calculator"""
from calculators.breadth_calculator import (
_score_200dma_breadth,
calculate_breadth_divergence,
)
class TestScore200dmaBreadth:
"""Boundary tests for 200DMA scoring."""
def test_critical_below_40(self):
assert _score_200dma_breadth(35) == 100
def test_healthy_above_70(self):
assert _score_200dma_breadth(75) == 5
def test_calibration_62_26(self):
"""62.26% breadth should score ~24 (healthy)."""
score = _score_200dma_breadth(62.26)
assert 18 <= score <= 30 # Reasonable range around 24
def test_boundary_50(self):
score = _score_200dma_breadth(50)
assert score == 55
def test_boundary_60(self):
score = _score_200dma_breadth(60)
assert score == 30
class TestCalculateBreadthDivergence:
"""Integration tests."""
def test_missing_breadth_returns_50(self):
"""None 200DMA → score 50, data_available=False."""
result = calculate_breadth_divergence(None, None, -2.0)
assert result["score"] == 50
assert result["data_available"] is False
def test_available_data_has_flag(self):
"""Valid data → data_available=True."""
result = calculate_breadth_divergence(65.0, 55.0, -1.0)
assert result["data_available"] is True
def test_near_highs_divergence(self):
"""Index near highs + weak breadth → high score."""
result = calculate_breadth_divergence(45.0, 30.0, -2.0)
assert result["score"] >= 70
assert result["divergence_detected"] is True
def test_not_near_highs_halved(self):
"""Index NOT near highs → score halved."""
near = calculate_breadth_divergence(45.0, None, -2.0)
far = calculate_breadth_divergence(45.0, None, -8.0)
assert far["score"] < near["score"]
def test_50dma_supplement_boost(self):
"""Low 50DMA breadth at highs adds +10."""
base = calculate_breadth_divergence(55.0, 80.0, -2.0)
boosted = calculate_breadth_divergence(55.0, 25.0, -2.0)
assert boosted["score"] > base["score"]
def test_auto_fetched_breadth_source_marker(self):
"""breadth_source can be set on comp4 result after calculate."""
result = calculate_breadth_divergence(64.81, None, -1.0)
result["breadth_source"] = "auto"
result["breadth_auto_date"] = "2026-02-18"
assert result["breadth_source"] == "auto"
assert result["breadth_auto_date"] == "2026-02-18"
assert result["data_available"] is True
scripts/tests/test_defensive_rotation.py
"""Tests for Defensive Rotation Calculator"""
from calculators.defensive_rotation_calculator import (
_score_rotation,
calculate_defensive_rotation,
)
class TestScoreRotation:
"""Boundary tests for rotation scoring."""
def test_growth_leading(self):
"""Negative relative = growth leading = healthy."""
assert _score_rotation(-3.0) == 0
assert _score_rotation(-1.0) <= 15
def test_slight_defensive_tilt(self):
assert 20 <= _score_rotation(0.3) <= 40
def test_strong_defensive_rotation(self):
assert _score_rotation(5.0) == 100
def test_moderate_rotation(self):
assert 60 <= _score_rotation(2.0) <= 80
class TestCalculateDefensiveRotation:
"""Integration tests."""
def test_missing_data_returns_50(self):
"""No ETF data -> score 50 (neutral), data_available=False."""
result = calculate_defensive_rotation({})
assert result["score"] == 50
assert result["data_available"] is False
def test_partial_data_still_computes(self):
"""At least 1 defensive + 1 offensive ETF should compute."""
historical = {
"XLU": [{"close": 70 + i * 0.1, "volume": 500000} for i in range(55)],
"XLK": [{"close": 200 - i * 0.5, "volume": 2000000} for i in range(55)],
}
result = calculate_defensive_rotation(historical)
assert result["data_available"] is False # 2/8 = 25% < 75%
assert "score" in result
def test_growth_leading_scenario(self):
"""Offensive outperforming defensive -> low score."""
historical = {}
for sym in ["XLU", "XLP", "XLV", "VNQ"]:
historical[sym] = [{"close": 50, "volume": 500000} for _ in range(55)]
for sym in ["XLK", "XLC", "XLY", "QQQ"]:
historical[sym] = [{"close": 105 - i * 0.25, "volume": 2000000} for i in range(55)]
result = calculate_defensive_rotation(historical)
assert result["score"] <= 20
class TestFetchSuccessRate:
"""Test fetch success rate tracking."""
def test_all_fetched(self):
"""All 8 ETFs with data -> fetch_success_rate = 1.0."""
historical = {}
for sym in ["XLU", "XLP", "XLV", "VNQ", "XLK", "XLC", "XLY", "QQQ"]:
historical[sym] = [{"close": 100 + i * 0.1, "volume": 500000} for i in range(55)]
result = calculate_defensive_rotation(historical)
assert result["fetch_success_rate"] == 1.0
assert result["data_available"] is True
def test_partial_data_rate(self):
"""Only some ETFs have data."""
historical = {
"XLU": [{"close": 50, "volume": 500000} for _ in range(55)],
"XLK": [{"close": 200, "volume": 2000000} for _ in range(55)],
}
result = calculate_defensive_rotation(historical)
assert result["fetch_success_rate"] == 0.25 # 2/8
class TestMultiPeriodConfirmation:
"""Test multi-period rotation analysis."""
def test_confirmed_all_periods_defensive(self):
"""All 3 periods show defensive leading -> confirmed."""
historical = {}
# Defensive ETFs going up
for sym in ["XLU", "XLP", "XLV", "VNQ"]:
historical[sym] = [{"close": 110 - i * 0.2, "volume": 500000} for i in range(55)]
# Offensive ETFs going down
for sym in ["XLK", "XLC", "XLY", "QQQ"]:
historical[sym] = [{"close": 90 + i * 0.5, "volume": 2000000} for i in range(55)]
result = calculate_defensive_rotation(historical)
assert result["confirmation"] == "confirmed"
assert "multi_period" in result
assert len(result["multi_period"]) >= 2
def test_unconfirmed_mixed_periods(self):
"""Some periods defensive, some not -> unconfirmed, capped at 80."""
historical = {}
# Create a scenario where short-term is defensive but long-term is not
for sym in ["XLU", "XLP", "XLV", "VNQ"]:
# Recent 10 days up, older days flat
closes = [60 - i * 0.3 for i in range(15)] + [50 for _ in range(40)]
historical[sym] = [{"close": c, "volume": 500000} for c in closes]
for sym in ["XLK", "XLC", "XLY", "QQQ"]:
# Recent 10 days flat, older days up
closes = [50 for _ in range(15)] + [50 - i * 0.3 for i in range(40)]
historical[sym] = [{"close": c, "volume": 2000000} for c in closes]
result = calculate_defensive_rotation(historical)
# Score should be capped at 80 if unconfirmed
if result["confirmation"] == "unconfirmed":
assert result["score"] <= 80
def test_multi_period_details_present(self):
"""Result should include multi_period dict with period details."""
historical = {}
for sym in ["XLU", "XLP", "XLV", "VNQ", "XLK", "XLC", "XLY", "QQQ"]:
historical[sym] = [{"close": 100 + i * 0.1, "volume": 500000} for i in range(55)]
result = calculate_defensive_rotation(historical)
assert "multi_period" in result
# Should have at least 10d and 20d periods (40d needs 41+ days)
assert 10 in result["multi_period"] or 20 in result["multi_period"]
scripts/tests/test_delta.py
"""Tests for Delta Tracking (load_previous_report, compute_deltas)"""
import json
import os
import tempfile
from market_top_detector import _compute_deltas, _load_previous_report
class TestLoadPreviousReport:
"""Test loading previous JSON report."""
def test_no_files_returns_none(self):
with tempfile.TemporaryDirectory() as tmpdir:
result = _load_previous_report(tmpdir)
assert result is None
def test_loads_latest_file(self):
with tempfile.TemporaryDirectory() as tmpdir:
# Create two report files
for ts in ["2026-02-18_100000", "2026-02-19_100000"]:
path = os.path.join(tmpdir, f"market_top_{ts}.json")
with open(path, "w", encoding="utf-8") as f:
json.dump({"timestamp": ts}, f)
result = _load_previous_report(tmpdir)
assert result["timestamp"] == "2026-02-19_100000"
def test_invalid_json_returns_none(self):
with tempfile.TemporaryDirectory() as tmpdir:
path = os.path.join(tmpdir, "market_top_2026-02-19_100000.json")
with open(path, "w", encoding="utf-8") as f:
f.write("not json")
result = _load_previous_report(tmpdir)
assert result is None
class TestComputeDeltas:
"""Test delta computation."""
def test_no_previous_all_first_run(self):
scores = {
"distribution_days": 50,
"leading_stocks": 40,
"defensive_rotation": 30,
"breadth_divergence": 20,
"index_technical": 10,
"sentiment": 5,
}
result = _compute_deltas(scores, None)
for _key, info in result["components"].items():
assert info["direction"] == "first_run"
assert result["composite_direction"] == "first_run"
def test_worsening_detected(self):
scores = {
"distribution_days": 80,
"leading_stocks": 50,
"defensive_rotation": 50,
"breadth_divergence": 50,
"index_technical": 50,
"sentiment": 50,
}
prev = {
"components": {
"distribution_days": {"score": 60},
"leading_stocks": {"score": 50},
"defensive_rotation": {"score": 50},
"breadth_divergence": {"score": 50},
"index_technical": {"score": 50},
"sentiment": {"score": 50},
},
"composite": {"composite_score": 52},
"metadata": {"generated_at": "2026-02-18 10:00:00"},
}
result = _compute_deltas(scores, prev)
assert result["components"]["distribution_days"]["direction"] == "worsening"
assert result["components"]["distribution_days"]["delta"] == 20
def test_improving_detected(self):
scores = {
"distribution_days": 30,
"leading_stocks": 50,
"defensive_rotation": 50,
"breadth_divergence": 50,
"index_technical": 50,
"sentiment": 50,
}
prev = {
"components": {
"distribution_days": {"score": 60},
"leading_stocks": {"score": 50},
"defensive_rotation": {"score": 50},
"breadth_divergence": {"score": 50},
"index_technical": {"score": 50},
"sentiment": {"score": 50},
},
"composite": {"composite_score": 52},
"metadata": {"generated_at": "2026-02-18"},
}
result = _compute_deltas(scores, prev)
assert result["components"]["distribution_days"]["direction"] == "improving"
assert result["components"]["distribution_days"]["delta"] == -30
def test_stable_within_3(self):
scores = {
"distribution_days": 52,
"leading_stocks": 50,
"defensive_rotation": 50,
"breadth_divergence": 50,
"index_technical": 50,
"sentiment": 50,
}
prev = {
"components": {
"distribution_days": {"score": 50},
"leading_stocks": {"score": 50},
"defensive_rotation": {"score": 50},
"breadth_divergence": {"score": 50},
"index_technical": {"score": 50},
"sentiment": {"score": 50},
},
"composite": {"composite_score": 50},
"metadata": {"generated_at": "2026-02-18"},
}
result = _compute_deltas(scores, prev)
assert result["components"]["distribution_days"]["direction"] == "stable"
scripts/tests/test_distribution_day.py
"""Tests for Distribution Day Calculator"""
from calculators.distribution_day_calculator import (
_calculate_clustering_factor,
_count_distribution_days,
_score_distribution_days,
calculate_distribution_days,
)
from helpers import make_history, make_history_with_volumes
class TestScoreDistributionDays:
"""Boundary tests for scoring thresholds."""
def test_zero_days(self):
assert _score_distribution_days(0) == 0
def test_one_day(self):
assert _score_distribution_days(1) == 15
def test_two_days(self):
assert _score_distribution_days(2) == 30
def test_three_days(self):
assert _score_distribution_days(3) == 55
def test_four_days_oneil_threshold(self):
assert _score_distribution_days(4) == 75
def test_five_days(self):
assert _score_distribution_days(5) == 90
def test_six_plus_days(self):
assert _score_distribution_days(6) == 100
assert _score_distribution_days(8) == 100
def test_fractional_stalling(self):
"""Stalling days count as 0.5"""
assert _score_distribution_days(3.5) == 55 # 3.5 >= 3
assert _score_distribution_days(4.5) == 75 # 4.5 >= 4
class TestCountDistributionDays:
"""Tests for distribution day detection logic."""
def test_empty_history(self):
result = _count_distribution_days([], "TEST")
assert result["distribution_days"] == 0
def test_single_day(self):
result = _count_distribution_days([{"close": 100, "volume": 1000}], "TEST")
assert result["distribution_days"] == 0
def test_distribution_day_detected(self):
"""Price drop >= 0.2% on higher volume = distribution day."""
history = make_history_with_volumes(
[
(99.0, 1200000), # Today: -1% drop, higher volume -> distribution
(100.0, 1000000), # Yesterday
(100.5, 800000), # Two days ago: lower volume than yesterday
]
)
result = _count_distribution_days(history, "TEST")
assert result["distribution_days"] >= 1
def test_stalling_day_detected(self):
"""Volume increases but price gain < 0.1% = stalling day."""
history = make_history_with_volumes(
[
(100.05, 1200000), # Today: +0.05%, higher volume
(100.0, 1000000), # Yesterday
(99.5, 900000), # Two days ago
]
)
result = _count_distribution_days(history, "TEST")
assert result["stalling_days"] >= 1
def test_window_index_in_details(self):
"""Distribution day details should include window_index."""
history = make_history_with_volumes(
[
(99.0, 1200000),
(100.0, 1000000),
(100.5, 800000),
]
)
result = _count_distribution_days(history, "TEST")
for d in result["details"]:
assert "window_index" in d
class TestClusteringFactor:
"""Test clustering factor calculation."""
def test_no_events(self):
result = _calculate_clustering_factor([])
assert result["factor"] == 0.0
assert result["total_count"] == 0
def test_all_recent(self):
"""All events in recent 5 days -> factor = 1.0."""
details = [
{"window_index": 0, "type": "distribution"},
{"window_index": 2, "type": "distribution"},
{"window_index": 4, "type": "distribution"},
]
result = _calculate_clustering_factor(details)
assert result["factor"] == 1.0
assert result["recent_count"] == 3
def test_spread_evenly(self):
"""Events spread across window -> low factor."""
details = [
{"window_index": 0, "type": "distribution"},
{"window_index": 10, "type": "distribution"},
{"window_index": 20, "type": "distribution"},
]
result = _calculate_clustering_factor(details)
assert result["factor"] < 0.5
def test_mixed(self):
"""Some recent, some old."""
details = [
{"window_index": 1, "type": "distribution"},
{"window_index": 3, "type": "distribution"},
{"window_index": 12, "type": "distribution"},
{"window_index": 18, "type": "distribution"},
]
result = _calculate_clustering_factor(details)
assert result["factor"] == 0.5 # 2/4
class TestClusteringMultiplier:
"""Test that clustering boosts the score."""
def test_clustered_boosts_score(self):
"""Clustered distribution days should get 1.15x multiplier."""
# Create history with 3 distribution days all in the last 5 days
# Days 0-4: distribution days (price drops on higher volume)
cv = []
for i in range(26):
if i < 3:
cv.append((99.0 - i * 0.5, 1200000 + i * 100000)) # Drop, high vol
elif i == 3:
cv.append((100.0, 1000000)) # Base
else:
cv.append((100.0 + (i - 3) * 0.1, 900000)) # Normal
sp_hist = make_history_with_volumes(cv)
flat_hist = make_history([100] * 26)
result = calculate_distribution_days(sp_hist, flat_hist)
if result["effective_count"] >= 2 and result["clustering"]["factor"] > 0.5:
assert result["clustering_applied"] is True
assert result["score"] >= result["raw_score"]
def test_spread_no_boost(self):
"""Spread-out distribution days should NOT get multiplier."""
# If no clustering, raw_score == score
flat_hist = make_history([100] * 30)
result = calculate_distribution_days(flat_hist, flat_hist)
assert result["clustering_applied"] is False
assert result["score"] == result["raw_score"]
class TestCalculateDistributionDays:
"""Integration tests."""
def test_uses_higher_count(self):
"""Should use the worse of S&P 500 / NASDAQ."""
sp_history = make_history([100] * 30)
nasdaq_cv = [(99.0 - i * 0.3, 1200000 + i * 10000) for i in range(15)]
nasdaq_cv += [(100.0 + i * 0.1, 1000000) for i in range(15)]
nasdaq_history = make_history_with_volumes(nasdaq_cv)
result = calculate_distribution_days(sp_history, nasdaq_history)
assert "score" in result
assert "effective_count" in result
assert "clustering" in result
assert "raw_score" in result
scripts/tests/test_fmp_client.py
"""Tests for FMP Client VIX term structure auto-detection and endpoint fallback"""
import os
import sys
from unittest.mock import MagicMock, Mock, patch
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
class TestVixTermStructure:
"""Test VIX term structure auto-classification."""
def _make_client(self):
"""Create a mock FMPClient without real API key."""
with patch.dict(os.environ, {"FMP_API_KEY": "test_key"}):
from fmp_client import FMPClient
client = FMPClient(api_key="test_key")
return client
def test_steep_contango(self):
"""VIX/VIX3M < 0.85 -> steep_contango."""
client = self._make_client()
client.get_quote = MagicMock(
side_effect=lambda s: [{"price": 12.0}] if "VIX3M" not in s else [{"price": 16.0}]
)
result = client.get_vix_term_structure()
assert result is not None
assert result["classification"] == "steep_contango"
assert result["ratio"] < 0.85
def test_contango(self):
"""VIX/VIX3M 0.85-0.95 -> contango."""
client = self._make_client()
client.get_quote = MagicMock(
side_effect=lambda s: [{"price": 14.0}] if "VIX3M" not in s else [{"price": 15.5}]
)
result = client.get_vix_term_structure()
assert result["classification"] == "contango"
def test_flat(self):
"""VIX/VIX3M 0.95-1.05 -> flat."""
client = self._make_client()
client.get_quote = MagicMock(
side_effect=lambda s: [{"price": 15.0}] if "VIX3M" not in s else [{"price": 15.2}]
)
result = client.get_vix_term_structure()
assert result["classification"] == "flat"
def test_backwardation(self):
"""VIX/VIX3M > 1.05 -> backwardation."""
client = self._make_client()
client.get_quote = MagicMock(
side_effect=lambda s: [{"price": 22.0}] if "VIX3M" not in s else [{"price": 18.0}]
)
result = client.get_vix_term_structure()
assert result["classification"] == "backwardation"
def test_unavailable(self):
"""VIX3M unavailable -> None."""
client = self._make_client()
client.get_quote = MagicMock(
side_effect=lambda s: [{"price": 15.0}] if "VIX3M" not in s else None
)
result = client.get_vix_term_structure()
assert result is None
def _mock_response(status_code, json_data=None, text=""):
"""Create a mock requests.Response."""
resp = Mock()
resp.status_code = status_code
resp.json = Mock(return_value=json_data)
resp.text = text
return resp
class TestEndpointFallback:
"""Test stable/v3 endpoint fallback logic in FMPClient."""
def _make_client(self):
"""Create FMPClient with mocked session."""
with patch.dict(os.environ, {"FMP_API_KEY": "test_key"}):
from fmp_client import FMPClient
client = FMPClient(api_key="test_key")
client.RATE_LIMIT_DELAY = 0 # disable rate limiting in tests
return client
# ------------------------------------------------------------------
# Tier A — Fallback logic (4 tests)
# ------------------------------------------------------------------
def test_quote_stable_success(self):
"""Stable 200 returns data, v3 is NOT called."""
client = self._make_client()
quote_data = [{"symbol": "^GSPC", "price": 5500.0}]
call_log = []
def mock_get(url, params=None, timeout=None):
call_log.append(url)
if "stable" in url:
return _mock_response(200, quote_data)
return _mock_response(403, text="Forbidden")
client.session.get = mock_get
result = client.get_quote("^GSPC")
assert result == quote_data
assert len(call_log) == 1
assert "stable" in call_log[0]
def test_quote_stable_403_falls_back_to_v3(self):
"""Stable 403, v3 200 returns v3 data."""
client = self._make_client()
v3_data = [{"symbol": "^GSPC", "price": 5500.0}]
def mock_get(url, params=None, timeout=None):
if "stable" in url:
return _mock_response(403, text="Forbidden")
return _mock_response(200, v3_data)
client.session.get = mock_get
result = client.get_quote("^GSPC")
assert result == v3_data
def test_quote_both_fail(self):
"""Both FMP endpoints 403 and yfinance unavailable returns None."""
client = self._make_client()
def mock_get(url, params=None, timeout=None):
return _mock_response(403, text="Forbidden")
client.session.get = mock_get
# Simulate yfinance fallback also unavailable so the "all sources
# failed -> None" contract still holds.
client._get_hist_from_yfinance = lambda *a, **k: None
result = client.get_quote("^GSPC")
assert result is None
def test_historical_fallback_to_v3(self):
"""Stable 403, v3 200 returns v3 data for historical."""
client = self._make_client()
v3_data = {"symbol": "^GSPC", "historical": [{"date": "2026-03-20", "close": 5500.0}]}
def mock_get(url, params=None, timeout=None):
if "stable" in url:
return _mock_response(403, text="Forbidden")
return _mock_response(200, v3_data)
client.session.get = mock_get
result = client.get_historical_prices("^GSPC", days=80)
assert result == v3_data
assert "historical" in result
# ------------------------------------------------------------------
# Tier B — Response normalization (4 tests)
# ------------------------------------------------------------------
def test_historical_stable_v3_format_passthrough(self):
"""Stable returns v3-compatible format {'historical': [...]} — returned as-is."""
client = self._make_client()
data = {"symbol": "^GSPC", "historical": [{"date": "2026-03-20", "close": 5500.0}]}
def mock_get(url, params=None, timeout=None):
return _mock_response(200, data)
client.session.get = mock_get
result = client.get_historical_prices("^GSPC", days=80)
assert result == data
def test_historical_stable_batch_format_exact_match(self):
"""Stable returns historicalStockList with matching symbol — normalized."""
client = self._make_client()
batch_data = {
"historicalStockList": [
{
"symbol": "^GSPC",
"historical": [{"date": "2026-03-20", "close": 5500.0}],
}
]
}
def mock_get(url, params=None, timeout=None):
return _mock_response(200, batch_data)
client.session.get = mock_get
result = client.get_historical_prices("^GSPC", days=80)
assert result is not None
assert "historical" in result
assert result["historical"] == [{"date": "2026-03-20", "close": 5500.0}]
def test_historical_stable_batch_no_match_falls_back_to_v3(self):
"""Stable batch has wrong symbol, falls back to v3 which succeeds."""
client = self._make_client()
batch_data = {
"historicalStockList": [
{
"symbol": "SPY",
"historical": [{"date": "2026-03-20", "close": 550.0}],
}
]
}
v3_data = {"symbol": "^GSPC", "historical": [{"date": "2026-03-20", "close": 5500.0}]}
def mock_get(url, params=None, timeout=None):
if "stable" in url:
return _mock_response(200, batch_data)
return _mock_response(200, v3_data)
client.session.get = mock_get
result = client.get_historical_prices("^GSPC", days=80)
assert result == v3_data
def test_historical_batch_no_match_returns_none_when_v3_also_fails(self):
"""Stable batch no match + v3 403 + yfinance unavailable returns None."""
client = self._make_client()
batch_data = {
"historicalStockList": [
{
"symbol": "SPY",
"historical": [{"date": "2026-03-20", "close": 550.0}],
}
]
}
def mock_get(url, params=None, timeout=None):
if "stable" in url:
return _mock_response(200, batch_data)
return _mock_response(403, text="Forbidden")
client.session.get = mock_get
# Simulate yfinance fallback also unavailable.
client._get_hist_from_yfinance = lambda *a, **k: None
result = client.get_historical_prices("^GSPC", days=80)
assert result is None
# ------------------------------------------------------------------
# Tier B+ — Shape validation (2 tests)
# ------------------------------------------------------------------
def test_quote_rejects_non_list_response(self):
"""Stable returns truthy dict — skipped, falls back to v3."""
client = self._make_client()
error_dict = {"Error Message": "Invalid API KEY."}
v3_data = [{"symbol": "^GSPC", "price": 5500.0}]
def mock_get(url, params=None, timeout=None):
if "stable" in url:
return _mock_response(200, error_dict)
return _mock_response(200, v3_data)
client.session.get = mock_get
result = client.get_quote("^GSPC")
assert result == v3_data
def test_historical_rejects_non_dict_response(self):
"""Stable returns truthy list — skipped, falls back to v3."""
client = self._make_client()
bad_data = [1, 2, 3]
v3_data = {"symbol": "^GSPC", "historical": [{"date": "2026-03-20", "close": 5500.0}]}
def mock_get(url, params=None, timeout=None):
if "stable" in url:
return _mock_response(200, bad_data)
return _mock_response(200, v3_data)
client.session.get = mock_get
result = client.get_historical_prices("^GSPC", days=80)
assert result == v3_data
# ------------------------------------------------------------------
# Symbol mismatch protection
# ------------------------------------------------------------------
def test_quote_symbol_mismatch_falls_back(self):
"""Single-symbol quote returning wrong symbol is rejected."""
client = self._make_client()
wrong = _mock_response(200, [{"symbol": "SPY", "price": 500.0}])
correct = _mock_response(200, [{"symbol": "^GSPC", "price": 5000.0}])
client.session.get = MagicMock(side_effect=[wrong, correct])
result = client.get_quote("^GSPC")
assert result == [{"symbol": "^GSPC", "price": 5000.0}]
assert client.session.get.call_count == 2
def test_historical_symbol_mismatch_falls_back(self):
"""Single-symbol historical returning wrong symbol is rejected."""
client = self._make_client()
wrong = _mock_response(200, {"symbol": "SPY", "historical": [{"close": 500}]})
correct = _mock_response(200, {"symbol": "^GSPC", "historical": [{"close": 5000}]})
client.session.get = MagicMock(side_effect=[wrong, correct])
result = client.get_historical_prices("^GSPC", days=80)
assert result["symbol"] == "^GSPC"
assert client.session.get.call_count == 2
def test_batch_quote_skips_symbol_check(self):
"""Multi-symbol (batch) quote does not apply symbol mismatch check."""
client = self._make_client()
batch_data = [{"symbol": "^GSPC", "price": 5000}, {"symbol": "^VIX", "price": 20}]
resp = _mock_response(200, batch_data)
client.session.get = MagicMock(return_value=resp)
result = client.get_quote("^GSPC,^VIX")
assert result == batch_data
assert client.session.get.call_count == 1
# ------------------------------------------------------------------
# Skill-specific tests
# ------------------------------------------------------------------
def test_vix_term_structure_works_via_fallback(self):
"""VIX term structure succeeds when get_quote uses stable->v3 fallback."""
client = self._make_client()
vix_data = [{"symbol": "^VIX", "price": 18.0}]
vix3m_data = [{"symbol": "^VIX3M", "price": 20.0}]
def mock_get(url, params=None, timeout=None):
if "stable" in url:
return _mock_response(403, text="Forbidden")
# v3 fallback: route by symbol in URL path
if "^VIX3M" in url:
return _mock_response(200, vix3m_data)
if "^VIX" in url:
return _mock_response(200, vix_data)
return _mock_response(404, text="Not Found")
client.session.get = mock_get
result = client.get_vix_term_structure()
assert result is not None
assert result["vix"] == 18.0
assert result["vix3m"] == 20.0
# 18/20 = 0.9 -> contango
assert result["classification"] == "contango"
assert result["ratio"] == 0.9
def test_market_top_exits_on_sp500_failure(self):
"""main() exits with code 1 when S&P 500 data unavailable.
NOTE: patches `market_top_detector.FMPClient` (the symbol AS USED by
main()), not the test-file-local FMPClient. When pytest runs multiple
detector skills in the same session, conftest evicts and re-imports
`fmp_client` on skill switch, which produces multiple class objects
from the same source file. Patching the test-local FMPClient would
miss the class that main() actually uses.
"""
with patch("sys.argv", ["market_top_detector.py"]):
import market_top_detector
def fake_init(self, **kw):
self.api_key = "test" # pragma: allowlist secret
self.session = MagicMock()
self.cache = {}
self.last_call_time = 0
self.rate_limit_reached = False
self.retry_count = 0
self.max_retries = 1
self.api_calls_made = 0
with (
patch.object(market_top_detector.FMPClient, "__init__", fake_init),
patch.object(market_top_detector.FMPClient, "get_quote", return_value=None),
patch.object(
market_top_detector.FMPClient, "get_historical_prices", return_value=None
),
patch.object(
market_top_detector.FMPClient,
"get_api_stats",
return_value={
"cache_entries": 0,
"api_calls_made": 0,
"rate_limit_reached": False,
},
),
):
with pytest.raises(SystemExit) as exc_info:
market_top_detector.main()
assert exc_info.value.code == 1
def test_market_top_continues_on_vix_failure(self, capsys):
"""main() continues with warning when VIX unavailable (non-fatal)."""
sp500_quote = [{"symbol": "^GSPC", "price": 5500.0, "yearHigh": 5600.0}]
sp500_hist = {
"symbol": "^GSPC",
"historical": [
{
"date": f"2026-03-{20 - i:02d}",
"close": 5500.0 - i * 10,
"volume": 3_000_000_000,
"open": 5490.0 - i * 10,
"high": 5510.0 - i * 10,
"low": 5480.0 - i * 10,
}
for i in range(260)
],
}
qqq_quote = [{"symbol": "QQQ", "price": 480.0, "yearHigh": 500.0}]
qqq_hist = {
"symbol": "QQQ",
"historical": [
{
"date": f"2026-03-{20 - i:02d}",
"close": 480.0 - i,
"volume": 50_000_000,
"open": 479.0 - i,
"high": 481.0 - i,
"low": 478.0 - i,
}
for i in range(260)
],
}
def mock_quote(symbols):
if symbols == "^GSPC":
return sp500_quote
if symbols == "QQQ":
return qqq_quote
if symbols == "^VIX":
return None # VIX fails
if symbols == "^VIX3M":
return None
# Return generic quote for ETFs
return [
{
"symbol": symbols.split(",")[0],
"price": 100.0,
"yearHigh": 110.0,
"changesPercentage": 0.5,
"volume": 1_000_000,
"avgVolume": 1_000_000,
}
]
def mock_hist(symbol, days=365):
if symbol == "^GSPC":
return sp500_hist
if symbol == "QQQ":
return qqq_hist
# Generic history for ETFs
return {
"symbol": symbol,
"historical": [
{
"date": f"2026-03-{20 - i:02d}",
"close": 100.0 + i * 0.1,
"volume": 1_000_000,
"open": 99.9 + i * 0.1,
"high": 100.1 + i * 0.1,
"low": 99.8 + i * 0.1,
}
for i in range(min(days, 260))
],
}
def mock_batch_quotes(symbols):
result = {}
for s in symbols:
q = mock_quote(s)
if q:
for item in q:
result[item.get("symbol", s)] = item
return result
def mock_batch_hist(symbols, days=50):
result = {}
for s in symbols:
d = mock_hist(s, days)
if d and "historical" in d:
result[s] = d["historical"]
return result
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
with (
patch.dict(os.environ, {"FMP_API_KEY": "test_key"}), # pragma: allowlist secret
patch(
"sys.argv",
[
"market_top_detector.py",
"--static-basket",
"--no-auto-breadth",
"--breadth-200dma",
"60.0",
"--put-call",
"0.70",
"--vix-term",
"contango",
"--output-dir",
tmpdir,
],
),
):
import importlib
import market_top_detector
importlib.reload(market_top_detector)
# Patch market_top_detector.FMPClient (the symbol used by main())
# rather than a test-file-local fmp_client import. See
# test_market_top_exits_on_sp500_failure docstring for the
# cross-skill conftest eviction context.
with (
patch.object(
market_top_detector.FMPClient, "get_quote", side_effect=mock_quote
),
patch.object(
market_top_detector.FMPClient,
"get_historical_prices",
side_effect=mock_hist,
),
patch.object(
market_top_detector.FMPClient,
"get_batch_quotes",
side_effect=mock_batch_quotes,
),
patch.object(
market_top_detector.FMPClient,
"get_batch_historical",
side_effect=mock_batch_hist,
),
):
# Should NOT raise SystemExit — VIX failure is non-fatal
try:
market_top_detector.main()
except SystemExit:
pytest.fail("main() should not exit when only VIX fails")
captured = capsys.readouterr()
assert "WARN" in captured.out or "VIX unavailable" in captured.out
class TestEODFlatListSuccess:
"""Issue #64: stable EOD flat list -> public method success (regression)."""
def _make_client(self):
with patch.dict(os.environ, {"FMP_API_KEY": "test_key"}):
from fmp_client import FMPClient
client = FMPClient(api_key="test_key")
client.max_retries = 0
return client
@patch("fmp_client.requests.Session")
def test_get_historical_prices_normalizes_flat_list(self, mock_session_class):
"""Flat list response from new EOD endpoint -> dict contract preserved."""
mock_session = MagicMock()
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = [
{
"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_resp.text = ""
mock_session.get.return_value = mock_resp
mock_session_class.return_value = mock_session
client = self._make_client()
client.session = mock_session
result = client.get_historical_prices("SPY", days=2)
assert isinstance(result, dict), f"expected dict, got {type(result).__name__}"
assert result["symbol"] == "SPY"
assert len(result["historical"]) == 2
assert result["historical"][0]["close"] == 501.0
# URL regression: must hit /historical-price-eod/full with from/to
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 "from" in params and "to" in params
assert "timeseries" not in params
# --- yfinance fallback for paid-tier-gated symbols (indices/ETFs) ---
class _FakeColumns:
"""Plain columns object WITHOUT a `levels` attr (no MultiIndex)."""
class _FakeTS:
def __init__(self, iso):
self._iso = iso
def strftime(self, fmt):
return self._iso # tests only use %Y-%m-%d
class _FakeDF:
"""Minimal stand-in for a yfinance DataFrame (ascending by date)."""
def __init__(self, rows):
self._rows = rows # list of (Timestamp-like, dict)
self.empty = len(rows) == 0
self.columns = _FakeColumns()
def iterrows(self):
return iter(self._rows)
def _row(iso, o, h, lo, c, v):
return (_FakeTS(iso), {"Open": o, "High": h, "Low": lo, "Close": c, "Volume": v})
def _make_client():
with patch.dict(os.environ, {"FMP_API_KEY": "test_key"}):
from fmp_client import FMPClient
return FMPClient(api_key="test_key")
class TestYFinanceFallback:
"""get_historical_prices / get_quote fall back to yfinance when FMP returns
nothing (e.g. ^GSPC/^VIX or sector ETFs gated behind a paid FMP plan)."""
@patch("fmp_client.FMPClient._request_with_fallback", return_value=None)
def test_historical_fallback_invoked_and_shape(self, _mock_fmp):
client = _make_client()
fake_df = _FakeDF(
[
_row("2026-04-27", 1.0, 2.0, 0.5, 1.5, 100),
_row("2026-04-28", 1.5, 2.5, 1.0, 2.0, 200),
_row("2026-04-29", 2.0, 3.0, 1.5, 2.5, 300),
]
)
fake_yf = MagicMock()
fake_yf.download.return_value = fake_df
with patch.dict("sys.modules", {"yfinance": fake_yf}):
result = client.get_historical_prices("^GSPC", days=10)
assert isinstance(result, dict)
assert result["symbol"] == "^GSPC"
hist = result["historical"]
assert len(hist) == 3
# Most-recent-first (descending) per FMP contract
assert hist[0]["date"] == "2026-04-29"
assert hist[-1]["date"] == "2026-04-27"
for key in ("date", "open", "high", "low", "close", "adjClose", "volume"):
assert key in hist[0]
assert hist[0]["close"] == hist[0]["adjClose"] == 2.5
@patch("fmp_client.FMPClient._request_with_fallback", return_value=None)
def test_quote_fallback_derives_fields(self, _mock_fmp):
client = _make_client()
# ascending; latest close 2.5, prev close 2.0 -> +25%
fake_df = _FakeDF(
[
_row("2026-04-27", 1.0, 4.0, 0.5, 1.5, 100), # yearHigh source
_row("2026-04-28", 1.5, 2.5, 0.2, 2.0, 200), # yearLow source
_row("2026-04-29", 2.0, 3.0, 1.5, 2.5, 300),
]
)
fake_yf = MagicMock()
fake_yf.download.return_value = fake_df
with patch.dict("sys.modules", {"yfinance": fake_yf}):
quotes = client.get_quote("^VIX")
assert isinstance(quotes, list) and len(quotes) == 1
q = quotes[0]
assert q["symbol"] == "^VIX"
assert q["price"] == 2.5
assert q["previousClose"] == 2.0
assert q["change"] == 0.5
assert q["changesPercentage"] == 25.0
assert q["yearHigh"] == 4.0
assert q["yearLow"] == 0.2
@patch("fmp_client.FMPClient._request_with_fallback", return_value=None)
def test_empty_df_returns_none_and_not_cached(self, _mock_fmp):
client = _make_client()
fake_yf = MagicMock()
fake_yf.download.return_value = _FakeDF([])
with patch.dict("sys.modules", {"yfinance": fake_yf}):
result = client.get_historical_prices("^IXIC", days=5)
assert result is None
assert "prices_^IXIC_5" not in client.cache
@patch("fmp_client.FMPClient._request_with_fallback")
def test_fmp_success_does_not_call_yfinance(self, mock_fmp):
mock_fmp.return_value = {
"symbol": "SPY",
"historical": [{"date": "2026-04-29", "close": 1.0}],
}
client = _make_client()
fake_yf = MagicMock()
with patch.dict("sys.modules", {"yfinance": fake_yf}):
result = client.get_historical_prices("SPY", days=5)
assert result["symbol"] == "SPY"
fake_yf.download.assert_not_called()
@patch("fmp_client.FMPClient._request_with_fallback", return_value=None)
def test_batch_quote_not_single_skips_fallback(self, _mock_fmp):
"""Comma-separated symbols are left to FMP (no single-symbol fallback)."""
client = _make_client()
fake_yf = MagicMock()
with patch.dict("sys.modules", {"yfinance": fake_yf}):
result = client.get_quote("AAA,BBB")
assert result is None
fake_yf.download.assert_not_called()
scripts/tests/test_freshness.py
"""Tests for Data Freshness Management"""
from datetime import date, timedelta
import pytest
from utils import count_business_days
def _expected_factor(biz_days: int) -> float:
"""Map business days to expected freshness factor."""
if biz_days <= 1:
return 1.0
elif biz_days <= 3:
return 0.95
elif biz_days <= 7:
return 0.85
else:
return 0.70
class TestDataFreshness:
"""Test data freshness computation."""
def test_today_returns_1(self):
"""Data from today -> freshness factor 1.0."""
from market_top_detector import compute_data_freshness
today = date.today().isoformat()
result = compute_data_freshness({"breadth_200dma_date": today})
assert result["breadth_200dma"]["factor"] == 1.0
def test_recent_data_factor(self):
"""Data from a few calendar days ago uses business day counting."""
from market_top_detector import compute_data_freshness
d = date.today() - timedelta(days=2)
result = compute_data_freshness({"breadth_200dma_date": d.isoformat()})
biz = count_business_days(d, date.today())
assert result["breadth_200dma"]["factor"] == _expected_factor(biz)
def test_week_old_data_factor(self):
"""Data from ~1 week ago uses business day counting."""
from market_top_detector import compute_data_freshness
d = date.today() - timedelta(days=5)
result = compute_data_freshness({"breadth_200dma_date": d.isoformat()})
biz = count_business_days(d, date.today())
assert result["breadth_200dma"]["factor"] == _expected_factor(biz)
def test_old_data_returns_070(self):
"""Data from many calendar days ago -> 0.70 (business days > 7)."""
from market_top_detector import compute_data_freshness
# 20 calendar days guarantees 14+ business days -> factor 0.70
d = (date.today() - timedelta(days=20)).isoformat()
result = compute_data_freshness({"breadth_200dma_date": d})
assert result["breadth_200dma"]["factor"] == 0.70
def test_no_date_returns_1(self):
"""No date provided -> assume fresh (1.0)."""
from market_top_detector import compute_data_freshness
result = compute_data_freshness({})
assert result["overall_confidence"] == 1.0
def test_no_value_returns_none(self):
"""Date given but no value -> entry should still compute."""
from market_top_detector import compute_data_freshness
d = date.today().isoformat()
result = compute_data_freshness({"put_call_date": d})
assert result["put_call"]["factor"] == 1.0
def test_overall_confidence_is_min(self):
"""Overall confidence = min of all provided factors."""
from market_top_detector import compute_data_freshness
today = date.today().isoformat()
# 20 calendar days -> 14+ business days -> factor 0.70
old = (date.today() - timedelta(days=20)).isoformat()
result = compute_data_freshness(
{
"breadth_200dma_date": today,
"put_call_date": old,
}
)
assert result["overall_confidence"] == 0.70
def test_future_date_returns_070(self):
"""Future date should be treated as anomaly with factor 0.70."""
from market_top_detector import compute_data_freshness
future = (date.today() + timedelta(days=5)).isoformat()
result = compute_data_freshness({"breadth_200dma_date": future})
assert result["breadth_200dma"]["factor"] == 0.70
assert result["breadth_200dma"]["age_days"] is None
def test_weekend_tolerance(self):
"""Friday data should still be fresh on Monday (1 business day).
Instead of mocking date, we test the underlying count_business_days
that compute_data_freshness now uses.
"""
from utils import count_business_days
friday = date(2026, 3, 13) # Friday
monday = date(2026, 3, 16) # Monday
biz_days = count_business_days(friday, monday)
# 1 business day -> factor would be 1.0 (<=1 threshold)
assert biz_days == 1
def test_exchange_holiday_is_not_counted(self):
"""Good Friday is not an XNYS freshness day."""
from market_top_detector import compute_data_freshness
result = compute_data_freshness(
{"breadth_200dma_date": "2026-04-02"},
as_of_date=date(2026, 4, 6),
)
assert result["breadth_200dma"]["age_days"] == 1
assert result["breadth_200dma"]["factor"] == 1.0
def test_explicit_as_of_rejects_future_source_date(self):
from market_top_detector import compute_data_freshness
result = compute_data_freshness(
{"put_call_date": "2026-04-07"},
as_of_date=date(2026, 4, 6),
)
assert result["put_call"] == {
"date": "2026-04-07",
"age_days": None,
"factor": 0.70,
}
def test_history_filter_removes_future_and_invalid_rows():
from market_top_detector import _history_on_or_before
rows = [
{"date": "2026-07-07", "close": 3},
{"date": "bad", "close": 2},
{"date": "2026-07-06", "close": 1},
]
assert _history_on_or_before(rows, date(2026, 7, 6)) == [rows[2]]
def test_batch_history_filter_protects_leading_and_sector_calculators():
from market_top_detector import _batch_histories_on_or_before
valid = {"date": "2026-04-06", "close": 100}
raw = {
"LEADER": [
{"date": "2026-04-07", "close": 999},
{"date": "invalid", "close": 888},
valid,
],
"DEFENSIVE": [{"date": "2026-04-08", "close": 777}],
}
filtered = _batch_histories_on_or_before(raw, date(2026, 4, 6))
assert filtered == {"LEADER": [valid]}
def test_future_dated_scored_cli_input_fails_before_live_fetch(capsys):
from market_top_detector import main
rc = main(
[
"--breadth-200dma",
"80",
"--breadth-200dma-date",
"2099-01-04",
]
)
captured = capsys.readouterr()
assert rc == 2
assert "future-dated scored input" in captured.err
assert captured.out == ""
@pytest.mark.parametrize("value", ["2026-7-6", "not-a-date"])
def test_as_of_parser_requires_strict_iso_date(value):
from market_top_detector import parse_arguments
with pytest.raises(SystemExit) as exc:
parse_arguments(["--as-of", value])
assert exc.value.code == 2
scripts/tests/test_historical_comparator.py
"""Tests for Historical Top Pattern Comparator"""
from historical_comparator import (
COMPONENT_KEYS,
HISTORICAL_TOPS,
_compute_ssd,
compare_to_historical,
)
class TestComputeSSD:
"""Test SSD calculation."""
def test_identical_scores_zero_ssd(self):
scores = {k: 50 for k in COMPONENT_KEYS}
assert _compute_ssd(scores, scores) == 0.0
def test_known_ssd(self):
a = {k: 0 for k in COMPONENT_KEYS}
b = {k: 10 for k in COMPONENT_KEYS}
# Weighted: sum(weight_i * 10^2) = 100 * sum(weights) = 100 * 1.0 = 100.0
assert _compute_ssd(a, b) == 100.0
def test_weighted_ssd_prioritizes_high_weight(self):
"""Distribution days (25%) difference should contribute more than sentiment (10%)."""
base = {k: 50 for k in COMPONENT_KEYS}
# Only distribution_days differs by 20
a = dict(base)
a["distribution_days"] = 70
ssd_dist = _compute_ssd(a, base)
# Only sentiment differs by 20
b = dict(base)
b["sentiment"] = 70
ssd_sent = _compute_ssd(b, base)
# distribution_days (0.25 * 400 = 100) > sentiment (0.10 * 400 = 40)
assert ssd_dist > ssd_sent
class TestCompareToHistorical:
"""Test comparison to historical patterns."""
def test_all_zeros_closest(self):
"""All zeros should be closest to 2018 (lowest overall scores)."""
scores = {k: 0 for k in COMPONENT_KEYS}
result = compare_to_historical(scores)
assert result["closest_match"] is None
assert "2018" in result["nearest_candidate"]
assert len(result["comparisons"]) == 4
def test_all_100_closest(self):
"""All 100 should be closest to 2000 (highest overall scores)."""
scores = {k: 100 for k in COMPONENT_KEYS}
result = compare_to_historical(scores)
assert result["closest_match"] is None
assert "2000" in result["nearest_candidate"]
def test_ssd_ordering(self):
"""Comparisons should be sorted by SSD ascending."""
scores = {k: 50 for k in COMPONENT_KEYS}
result = compare_to_historical(scores)
ssds = [c["ssd"] for c in result["comparisons"]]
assert ssds == sorted(ssds)
def test_narrative_with_differences(self):
"""Large differences should appear in narrative."""
scores = {k: 0 for k in COMPONENT_KEYS}
scores["distribution_days"] = 100 # Very different from any top
result = compare_to_historical(scores)
assert result["status"] == "no_comparable_episode"
def test_narrative_close_match(self):
"""Scores close to a pattern should produce 'closely matches' text."""
# Use exact 2018 scores
scores = dict(HISTORICAL_TOPS["2018 (Q4 Correction)"])
result = compare_to_historical(scores)
assert result["closest_ssd"] == 0.0
assert "closely matches" in result["narrative"]
def test_incomplete_and_invalid_evidence():
import json
base = dict(HISTORICAL_TOPS["2018 (Q4 Correction)"])
cases = [{}] + [
{**base, "sentiment": v} for v in [None, True, "55", float("nan"), float("inf"), -1, 101]
]
for scores in cases:
result = compare_to_historical(scores)
assert result["closest_match"] is None
assert result["closest_ssd"] is None
assert result["nearest_candidate"] is None
assert result["comparisons"] == []
json.dumps(result, allow_nan=False)
result = compare_to_historical(base, {"sentiment": False})
assert result["status"] == "no_comparable_episode"
assert "sentiment" in result["narrative"]
def test_unrounded_rms_boundary(monkeypatch):
import historical_comparator as comparator
monkeypatch.setattr(comparator, "HISTORICAL_TOPS", {"fixture": {k: 50 for k in COMPONENT_KEYS}})
for difference, expected in [
(15, "comparable_illustration"),
(15.00001, "no_comparable_episode"),
]:
result = compare_to_historical({k: 50 + difference for k in COMPONENT_KEYS})
assert result["status"] == expected
scripts/tests/test_index_technical.py
"""Tests for Index Technical Calculator"""
from calculators.index_technical_calculator import (
_evaluate_index,
calculate_index_technical,
)
from helpers import make_history
class TestEvaluateIndex:
"""Test single index evaluation."""
def test_insufficient_data(self):
"""Less than 21 days → data_available=False."""
result = _evaluate_index("TEST", [{"close": 100}] * 10)
assert result["data_available"] is False
assert result["raw_score"] == 0
def test_sufficient_data(self):
"""21+ days → data_available=True."""
history = make_history([100 - i * 0.1 for i in range(50)])
result = _evaluate_index("TEST", history)
assert result["data_available"] is True
def test_below_21ema_adds_8(self):
"""Price significantly below 21 EMA should trigger flag."""
# Create downtrend: recent prices well below earlier ones
closes = [90] * 5 + [100 + i * 0.5 for i in range(45)]
history = make_history(closes)
result = _evaluate_index("TEST", history)
assert any("21 EMA" in f for f in result["flags"])
class TestCalculateIndexTechnical:
"""Test composite index technical score."""
def test_both_available(self):
"""Both S&P and NASDAQ have data → average them."""
sp = make_history([100 - i * 0.1 for i in range(50)])
nq = make_history([200 - i * 0.2 for i in range(50)])
result = calculate_index_technical(sp, nq)
assert result["data_available"] is True
assert result["sp500"]["data_available"] is True
assert result["nasdaq"]["data_available"] is True
def test_nasdaq_missing_uses_sp_only(self):
"""NASDAQ insufficient data → use S&P 500 only, no halving."""
sp = make_history([100 - i * 0.1 for i in range(50)])
nq_short = [{"close": 200}] * 5 # Too short
calculate_index_technical(sp, sp) # Both = S&P (warm up)
result_sp_only = calculate_index_technical(sp, nq_short)
# S&P-only score should equal S&P's raw_score (not halved)
sp_raw = result_sp_only["sp500"]["raw_score"]
assert result_sp_only["score"] == round(min(100, max(0, sp_raw)))
assert result_sp_only["nasdaq"]["data_available"] is False
def test_both_missing_returns_50(self):
"""Both insufficient → score=50, data_available=False."""
short = [{"close": 100}] * 5
result = calculate_index_technical(short, short)
assert result["score"] == 50
assert result["data_available"] is False
assert "NO DATA" in result["signal"]
def test_sp_missing_nasdaq_available(self):
"""S&P insufficient, NASDAQ available → use NASDAQ only."""
sp_short = [{"close": 100}] * 5
nq = make_history([200 - i * 0.2 for i in range(50)])
result = calculate_index_technical(sp_short, nq)
nq_raw = result["nasdaq"]["raw_score"]
assert result["score"] == round(min(100, max(0, nq_raw)))
scripts/tests/test_leading_stock.py
"""Tests for Leading Stock Calculator"""
from calculators.leading_stock_calculator import (
CANDIDATE_POOL,
DEFAULT_LEADING_ETFS,
_evaluate_etf,
calculate_leading_stock_health,
select_dynamic_basket,
)
class TestEvaluateETF:
"""Test individual ETF evaluation."""
def test_near_highs_healthy(self):
quote = {"price": 100, "yearHigh": 102, "yearLow": 80}
history = [{"close": 100 - i * 0.1, "volume": 1000000} for i in range(60)]
result = _evaluate_etf("TEST", quote, history)
assert result["deterioration_score"] <= 20
def test_deep_correction(self):
quote = {"price": 70, "yearHigh": 100, "yearLow": 65}
history = [{"close": 70 + i * 0.3, "volume": 1000000} for i in range(60)]
result = _evaluate_etf("TEST", quote, history)
assert result["deterioration_score"] >= 30
class TestCalculateLeadingStockHealth:
"""Test composite leading stock calculation."""
def test_no_data_returns_50(self):
"""No quotes -> score 50 (neutral), data_available=False."""
result = calculate_leading_stock_health({}, {})
assert result["score"] == 50
assert result["etfs_evaluated"] == 0
assert result["data_available"] is False
def test_healthy_leaders(self):
quotes = {}
historical = {}
for sym in ["ARKK", "WCLD", "IGV"]:
quotes[sym] = {"price": 100, "yearHigh": 102, "yearLow": 80}
historical[sym] = [{"close": 100 - i * 0.05, "volume": 1000000} for i in range(60)]
result = calculate_leading_stock_health(quotes, historical)
assert result["score"] <= 30 # Healthy
assert result["data_available"] is True
def test_amplification_at_60pct(self):
"""60%+ ETFs deteriorating triggers 1.3x amplification."""
quotes = {}
historical = {}
for sym in ["ARKK", "WCLD", "IGV", "XBI", "SOXX", "SMH", "KWEB", "TAN"]:
quotes[sym] = {"price": 70, "yearHigh": 100, "yearLow": 60}
historical[sym] = [
{"close": 70 + i * 0.5, "high": 72 + i * 0.5, "volume": 1000000} for i in range(60)
]
result = calculate_leading_stock_health(quotes, historical)
assert result["amplified"] is True
class TestFetchSuccessRate:
"""Test fetch success rate tracking."""
def test_all_fetched(self):
"""All ETFs have quotes -> success rate = 1.0."""
quotes = {}
historical = {}
for sym in ["A", "B", "C"]:
quotes[sym] = {"price": 100, "yearHigh": 102, "yearLow": 80}
historical[sym] = [{"close": 100, "volume": 1000000} for _ in range(60)]
result = calculate_leading_stock_health(quotes, historical)
assert result["fetch_success_rate"] == 1.0
assert result["data_available"] is True
def test_partial_fetched(self):
"""Some ETFs missing quotes -> lower success rate."""
quotes = {"A": {"price": 100, "yearHigh": 102, "yearLow": 80}}
historical = {"A": [{"close": 100, "volume": 1000000} for _ in range(60)]}
# Only 1 out of 1 in quotes -> 1.0
result = calculate_leading_stock_health(quotes, historical)
assert result["fetch_success_rate"] == 1.0
def test_below_threshold_marks_unavailable(self):
"""Below 75% success rate -> data_available=False."""
# quotes has 4 keys but only 2 have actual data
quotes = {
"A": {"price": 100, "yearHigh": 102, "yearLow": 80},
"B": None,
"C": None,
"D": None,
}
historical = {"A": [{"close": 100, "volume": 1000000} for _ in range(60)]}
result = calculate_leading_stock_health(quotes, historical)
# 1 success out of 4 = 0.25 < 0.75
assert result["fetch_success_rate"] < 0.75
assert result["data_available"] is False
class TestSelectDynamicBasket:
"""Test dynamic basket selection."""
def test_selects_top_n_by_proximity(self):
"""Should select ETFs closest to 52-week high."""
quotes = {
"SMH": {"price": 98, "yearHigh": 100},
"SOXX": {"price": 90, "yearHigh": 100},
"IGV": {"price": 95, "yearHigh": 100},
"WCLD": {"price": 80, "yearHigh": 100},
"XBI": {"price": 70, "yearHigh": 100},
"ARKK": {"price": 60, "yearHigh": 100},
}
result = select_dynamic_basket(quotes, top_n=3)
assert len(result) == 3
assert result[0] == "SMH" # 98% proximity
assert result[1] == "IGV" # 95% proximity
assert result[2] == "SOXX" # 90% proximity
def test_fallback_when_too_few_candidates(self):
"""Fewer than 5 valid candidates -> fall back to default basket."""
quotes = {
"SMH": {"price": 100, "yearHigh": 100},
"SOXX": {"price": 90, "yearHigh": 100},
}
result = select_dynamic_basket(quotes)
assert result == list(DEFAULT_LEADING_ETFS)
def test_skips_invalid_quotes(self):
"""Symbols with price=0 or yearHigh=0 should be skipped."""
quotes = {}
for i, sym in enumerate(CANDIDATE_POOL[:8]):
quotes[sym] = {"price": 100 - i * 5, "yearHigh": 100}
# Add invalid entries
quotes["BAD1"] = {"price": 0, "yearHigh": 100}
quotes["BAD2"] = {"price": 50, "yearHigh": 0}
result = select_dynamic_basket(quotes, top_n=5)
assert len(result) == 5
assert "BAD1" not in result
assert "BAD2" not in result
def test_empty_quotes_returns_default(self):
"""Empty quotes dict -> default basket."""
result = select_dynamic_basket({})
assert result == list(DEFAULT_LEADING_ETFS)
class TestPartialDataTracking:
"""Test partial data detection (quote but insufficient history)."""
def test_quote_only_not_available(self):
"""8 ETFs with quotes but no history -> data_available=False, partial=8."""
symbols = ["A", "B", "C", "D", "E", "F", "G", "H"]
quotes = {s: {"price": 100, "yearHigh": 102, "yearLow": 80} for s in symbols}
historical = {} # No history at all
result = calculate_leading_stock_health(quotes, historical, etf_list=symbols)
assert result["data_available"] is False
assert result["partial_data_count"] == 8
assert result["fetch_success_rate"] == 0.0
def test_mixed_full_and_partial_above_threshold(self):
"""6 full + 2 partial out of 8 -> data_available=True (75%), partial=2."""
symbols = ["A", "B", "C", "D", "E", "F", "G", "H"]
quotes = {s: {"price": 100, "yearHigh": 102, "yearLow": 80} for s in symbols}
historical = {}
for s in symbols[:6]:
# Full history (50+ bars)
historical[s] = [{"close": 100 - i * 0.1, "volume": 1000000} for i in range(60)]
for s in symbols[6:]:
# Partial history (< 50 bars)
historical[s] = [{"close": 100 - i * 0.1, "volume": 1000000} for i in range(20)]
result = calculate_leading_stock_health(quotes, historical, etf_list=symbols)
assert result["data_available"] is True # 6/8 = 0.75
assert result["partial_data_count"] == 2
assert result["fetch_success_rate"] == 0.75
def test_mixed_full_and_partial_below_threshold(self):
"""5 full + 3 partial out of 8 -> data_available=False (62.5%), partial=3."""
symbols = ["A", "B", "C", "D", "E", "F", "G", "H"]
quotes = {s: {"price": 100, "yearHigh": 102, "yearLow": 80} for s in symbols}
historical = {}
for s in symbols[:5]:
historical[s] = [{"close": 100 - i * 0.1, "volume": 1000000} for i in range(60)]
for s in symbols[5:]:
historical[s] = [{"close": 100 - i * 0.1, "volume": 1000000} for i in range(20)]
result = calculate_leading_stock_health(quotes, historical, etf_list=symbols)
assert result["data_available"] is False # 5/8 = 0.625 < 0.75
assert result["partial_data_count"] == 3
def test_no_data_includes_partial_count(self):
"""No data case should also include partial_data_count."""
result = calculate_leading_stock_health({}, {})
assert "partial_data_count" in result
assert result["partial_data_count"] == 0
class TestBasketModeInResult:
"""Test basket_mode and basket fields in result."""
def test_static_basket_mode(self):
"""Passing DEFAULT_LEADING_ETFS as etf_list -> static mode."""
quotes = {"A": {"price": 100, "yearHigh": 102, "yearLow": 80}}
historical = {"A": [{"close": 100, "volume": 1000000} for _ in range(60)]}
result = calculate_leading_stock_health(
quotes, historical, etf_list=list(DEFAULT_LEADING_ETFS)
)
assert result["basket_mode"] == "static"
assert result["basket"] == list(DEFAULT_LEADING_ETFS)
def test_dynamic_basket_mode(self):
"""Passing custom etf_list -> dynamic mode."""
etfs = ["SMH", "SOXX", "IGV"]
quotes = {}
historical = {}
for sym in etfs:
quotes[sym] = {"price": 100, "yearHigh": 102, "yearLow": 80}
historical[sym] = [{"close": 100, "volume": 1000000} for _ in range(60)]
result = calculate_leading_stock_health(quotes, historical, etf_list=etfs)
assert result["basket_mode"] == "dynamic"
assert result["basket"] == etfs
scripts/tests/test_market_calendar_contract.py
# GENERATED by scripts/generate_market_calendar.py — do not edit.
# Source of truth: scripts/market_calendar/ and config/market-calendar-dependency.yaml.
"""Generated contract tests for the vendored exchange calendar."""
from datetime import date, datetime
from zoneinfo import ZoneInfo
import _market_calendar as calendar
import pytest
def test_supported_cash_equity_venues():
assert calendar.supported_venues() == ("XLON", "XNAS", "XNYS", "XTKS")
def test_jpx_close_cutover_and_lunch_break():
before = calendar.session_for_date("XTKS", date(2024, 11, 1))
after = calendar.session_for_date("XTKS", date(2024, 11, 5))
assert before.market_close.strftime("%H:%M") == "15:00"
assert after.market_close.strftime("%H:%M") == "15:30"
assert after.break_start.strftime("%H:%M") == "11:30"
assert after.break_end.strftime("%H:%M") == "12:30"
assert (
calendar.is_open_at("XTKS", datetime(2024, 11, 5, 12, 0, tzinfo=ZoneInfo("Asia/Tokyo")))
is False
)
def test_us_holiday_early_close_and_dst():
assert calendar.session_for_date("XNYS", date(2026, 4, 3)) is None
early = calendar.session_for_date("XNYS", date(2026, 11, 27))
assert early.market_close.strftime("%H:%M %z") == "13:00 -0500"
winter = calendar.session_for_date("XNYS", date(2026, 1, 5))
summer = calendar.session_for_date("XNAS", date(2026, 7, 6))
assert winter.market_open.utcoffset().total_seconds() == -5 * 3600
assert summer.market_open.utcoffset().total_seconds() == -4 * 3600
def test_lse_uses_dst_aware_timezone():
winter = calendar.session_for_date("XLON", date(2026, 1, 5))
summer = calendar.session_for_date("XLON", date(2026, 7, 6))
assert winter.market_open.tzname() == "GMT"
assert summer.market_open.tzname() == "BST"
def test_next_session_skips_weekend_and_holiday():
session = calendar.next_session_on_or_after("XNYS", date(2026, 7, 3))
assert session.session_date == date(2026, 7, 6)
later = calendar.next_session_after("XNYS", date(2026, 7, 6))
assert later.session_date == date(2026, 7, 7)
def test_previous_session_skips_weekend_and_holiday():
session = calendar.previous_session_before("XNYS", date(2026, 7, 6))
assert session.session_date == date(2026, 7, 2)
same = calendar.previous_session_on_or_before("XNYS", date(2026, 7, 6))
assert same.session_date == date(2026, 7, 6)
@pytest.mark.parametrize(
("include_start", "include_end", "expected"),
[(False, False, 0), (False, True, 1), (True, False, 1), (True, True, 2)],
)
def test_count_session_boundaries(include_start, include_end, expected):
assert (
calendar.count_sessions(
"XNYS",
date(2026, 7, 2),
date(2026, 7, 6),
include_start=include_start,
include_end=include_end,
)
== expected
)
def test_same_day_and_reverse_contracts():
assert (
calendar.count_sessions(
"XNYS",
date(2026, 7, 6),
date(2026, 7, 6),
include_start=True,
include_end=True,
)
== 1
)
with pytest.raises(ValueError, match="on or before"):
calendar.count_sessions(
"XNYS",
date(2026, 7, 7),
date(2026, 7, 6),
include_start=False,
include_end=True,
)
def test_naive_datetime_and_unknown_venue_fail_closed():
with pytest.raises(ValueError, match="timezone-aware"):
calendar.is_open_at("XNYS", datetime(2026, 7, 6, 10, 0))
with pytest.raises(ValueError, match="Unsupported venue"):
calendar.session_for_date("NOPE", date(2026, 7, 6))
def test_parse_aware_datetime_accepts_z_and_rejects_date_only():
parsed = calendar.parse_aware_datetime("2026-07-06T14:00:00Z")
assert parsed.utcoffset().total_seconds() == 0
with pytest.raises(ValueError, match="timezone-aware"):
calendar.parse_aware_datetime("2026-07-06")
def test_close_is_exclusive_and_current_session_rolls_forward():
at_close = datetime(2026, 11, 27, 13, 0, tzinfo=ZoneInfo("America/New_York"))
assert calendar.is_open_at("XNYS", at_close) is False
assert calendar.current_or_next_session("XNYS", at_close).session_date == date(2026, 11, 30)
def test_provider_failure_is_not_treated_as_a_holiday(monkeypatch):
def fail_provider(_venue):
raise calendar.CalendarUnavailableError("calendar unavailable")
monkeypatch.setattr(calendar, "_provider_calendar", fail_provider)
with pytest.raises(calendar.CalendarUnavailableError, match="unavailable"):
calendar.session_for_date("XNYS", date(2026, 7, 6))
def test_search_horizon_failure_is_explicit(monkeypatch):
class EmptySchedule:
empty = True
monkeypatch.setattr(
calendar,
"_schedule",
lambda venue, start, end: ("XNYS", "America/New_York", EmptySchedule()),
)
with pytest.raises(calendar.CalendarUnavailableError, match="No XNYS session"):
calendar.next_session_on_or_after("XNYS", date(2026, 7, 6), max_search_days=1)
scripts/tests/test_math_utils.py
"""Tests for EMA/SMA shared utilities"""
import pytest
from calculators.math_utils import calc_ema, calc_sma
class TestCalcEma:
"""Boundary and correctness tests for calc_ema."""
def test_empty_list_raises(self):
with pytest.raises(ValueError, match="prices must not be empty"):
calc_ema([], 10)
def test_period_zero_raises(self):
with pytest.raises(ValueError, match="period must be >= 1"):
calc_ema([100.0], 0)
def test_period_negative_raises(self):
with pytest.raises(ValueError, match="period must be >= 1"):
calc_ema([100.0], -1)
def test_period_one(self):
"""Period=1 EMA should approximate most recent price."""
prices = [100.0, 90.0, 80.0]
result = calc_ema(prices, 1)
# With k=1.0 for period=1, EMA should be the last chronological value = 100.0
assert abs(result - 100.0) < 0.01
def test_period_exceeds_length(self):
"""period > len(prices) → simple average of all."""
prices = [100.0, 110.0]
result = calc_ema(prices, 10)
assert abs(result - 105.0) < 0.01
def test_known_ema_value(self):
"""Verify EMA calculation with known values."""
# 5 prices (most recent first): 105, 104, 103, 102, 101
# Chronological: 101, 102, 103, 104, 105
# SMA(3) of first 3 = (101+102+103)/3 = 102.0
# k = 2/(3+1) = 0.5
# EMA after 104: 104*0.5 + 102*0.5 = 103.0
# EMA after 105: 105*0.5 + 103*0.5 = 104.0
prices = [105, 104, 103, 102, 101]
result = calc_ema(prices, 3)
assert abs(result - 104.0) < 0.01
class TestCalcSma:
"""Boundary and correctness tests for calc_sma."""
def test_empty_list_raises(self):
with pytest.raises(ValueError, match="prices must not be empty"):
calc_sma([], 10)
def test_period_zero_raises(self):
with pytest.raises(ValueError, match="period must be >= 1"):
calc_sma([100.0], 0)
def test_period_negative_raises(self):
with pytest.raises(ValueError, match="period must be >= 1"):
calc_sma([100.0], -1)
def test_period_one(self):
"""Period=1 SMA = most recent price."""
prices = [100.0, 90.0, 80.0]
result = calc_sma(prices, 1)
assert result == 100.0
def test_period_exceeds_length(self):
"""period > len(prices) → simple average of all."""
prices = [100.0, 110.0]
result = calc_sma(prices, 10)
assert abs(result - 105.0) < 0.01
def test_known_sma_value(self):
"""SMA of first N prices (most recent first)."""
prices = [105, 104, 103, 102, 101]
result = calc_sma(prices, 3)
assert abs(result - (105 + 104 + 103) / 3) < 0.01
scripts/tests/test_report_generator.py
"""Tests for Report Generator"""
import os
import tempfile
from report_generator import _delta_arrow, generate_markdown_report
class TestDeltaArrow:
"""Test delta arrow formatting."""
def test_first_run(self):
assert _delta_arrow({"direction": "first_run", "delta": 0}) == "---"
def test_worsening(self):
result = _delta_arrow({"direction": "worsening", "delta": 10})
assert "↑" in result
assert "+10" in result
def test_improving(self):
result = _delta_arrow({"direction": "improving", "delta": -15})
assert "↓" in result
assert "-15" in result
def test_stable(self):
result = _delta_arrow({"direction": "stable", "delta": 2})
assert result == "→"
def test_none_input(self):
assert _delta_arrow(None) == "---"
class TestMarkdownReportWithDelta:
"""Test Markdown report output with delta information."""
def _make_analysis(self, delta=None):
return {
"metadata": {"generated_at": "2026-02-19 10:00:00", "data_mode": "test"},
"composite": {
"composite_score": 45,
"zone": "Orange (Elevated Risk)",
"zone_color": "orange",
"risk_budget": "60-75%",
"guidance": "Test guidance",
"actions": ["Action 1"],
"strongest_warning": {
"label": "Distribution",
"score": 80,
"component": "distribution_days",
},
"weakest_warning": {"label": "Sentiment", "score": 10, "component": "sentiment"},
"data_quality": {
"label": "Complete (6/6)",
"available_count": 6,
"total_components": 6,
"missing_components": [],
},
"correlation_adjustment": [],
"component_scores": {
k: {
"score": 50,
"adjusted_score": 50,
"weight": w,
"weighted_contribution": 50 * w,
"label": k.replace("_", " ").title(),
}
for k, w in [
("distribution_days", 0.25),
("leading_stocks", 0.20),
("defensive_rotation", 0.15),
("breadth_divergence", 0.15),
("index_technical", 0.15),
("sentiment", 0.10),
]
},
},
"components": {
k: {"score": 50, "signal": "TEST"}
for k in [
"distribution_days",
"leading_stocks",
"defensive_rotation",
"breadth_divergence",
"index_technical",
"sentiment",
]
},
"follow_through_day": {"applicable": False},
"delta": delta,
}
def test_report_with_delta(self):
"""Report with delta should include Δ column."""
delta = {
"components": {
"distribution_days": {"delta": 10, "direction": "worsening", "previous": 40},
"leading_stocks": {"delta": -5, "direction": "improving", "previous": 55},
"defensive_rotation": {"delta": 0, "direction": "stable", "previous": 50},
"breadth_divergence": {"delta": 2, "direction": "stable", "previous": 48},
"index_technical": {"delta": -20, "direction": "improving", "previous": 70},
"sentiment": {"delta": 5, "direction": "worsening", "previous": 45},
},
"composite_delta": 3.5,
"composite_direction": "worsening",
"previous_date": "2026-02-18 10:00:00",
"previous_composite": 41.5,
}
analysis = self._make_analysis(delta=delta)
with tempfile.NamedTemporaryFile(suffix=".md", delete=False) as f:
path = f.name
try:
generate_markdown_report(analysis, path)
with open(path, encoding="utf-8") as f:
content = f.read()
assert "Δ" in content
assert "↑" in content
assert "vs. Previous Run" in content
finally:
os.unlink(path)
def test_report_without_delta(self):
"""Report without delta should not include Δ column."""
analysis = self._make_analysis(delta=None)
with tempfile.NamedTemporaryFile(suffix=".md", delete=False) as f:
path = f.name
try:
generate_markdown_report(analysis, path)
with open(path) as f:
content = f.read()
assert "| Δ |" not in content
finally:
os.unlink(path)
def test_report_with_none_strongest_weakest(self):
"""Report should handle None component for strongest/weakest."""
analysis = self._make_analysis()
# Override strongest/weakest to None
analysis["composite"]["strongest_warning"] = {
"component": None,
"label": "N/A",
"score": 0,
}
analysis["composite"]["weakest_warning"] = {
"component": None,
"label": "N/A",
"score": 0,
}
with tempfile.NamedTemporaryFile(suffix=".md", delete=False) as f:
path = f.name
try:
generate_markdown_report(analysis, path)
with open(path) as f:
content = f.read()
assert "N/A" in content
# Should not crash
finally:
os.unlink(path)
def test_report_with_partial_data_count(self):
"""Report should include partial data note when partial_data_count > 0."""
analysis = self._make_analysis()
analysis["components"]["leading_stocks"] = {
"score": 50,
"signal": "TEST",
"etfs_evaluated": 8,
"etfs_deteriorating": 3,
"deteriorating_pct": 37.5,
"amplified": False,
"partial_data_count": 2,
"etf_details": {},
}
with tempfile.NamedTemporaryFile(suffix=".md", delete=False) as f:
path = f.name
try:
generate_markdown_report(analysis, path)
with open(path) as f:
content = f.read()
assert "2 ETF(s) had insufficient history" in content
finally:
os.unlink(path)
def test_first_run_note(self):
"""First run delta should show note."""
delta = {
"components": {
k: {"delta": 0, "direction": "first_run"}
for k in [
"distribution_days",
"leading_stocks",
"defensive_rotation",
"breadth_divergence",
"index_technical",
"sentiment",
]
},
"composite_delta": 0,
"composite_direction": "first_run",
"previous_date": None,
}
analysis = self._make_analysis(delta=delta)
with tempfile.NamedTemporaryFile(suffix=".md", delete=False) as f:
path = f.name
try:
generate_markdown_report(analysis, path)
with open(path) as f:
content = f.read()
assert "First run" in content
finally:
os.unlink(path)
scripts/tests/test_risk_interpretation.py
"""Risk interpretation must survive JSON and Markdown report rendering."""
import json
from historical_comparator import COMPONENT_KEYS, compare_to_historical
from report_generator import generate_json_report, generate_markdown_report
from scorer import calculate_composite_score
def test_report_no_match_and_uncalibrated_score(tmp_path):
analysis = {
"composite": calculate_composite_score({key: 70 for key in COMPONENT_KEYS}),
"historical_comparison": compare_to_historical({key: 0 for key in COMPONENT_KEYS}),
}
json_path = tmp_path / "report.json"
md_path = tmp_path / "report.md"
generate_json_report(analysis, str(json_path))
generate_markdown_report(analysis, str(md_path))
saved = json.loads(json_path.read_text())
assert saved["composite"]["calibrated_probability"] is False
assert saved["composite"]["score_kind"] == "heuristic_risk_score"
assert saved["composite"]["zone"] == "Red (High Risk)"
assert saved["historical_comparison"]["closest_match"] is None
text = md_path.read_text()
assert "Heuristic Risk Score" in text
assert "not an estimated event probability" in text
assert "No comparable episode" in text
assert "Closest Pattern:** None" not in text
scripts/tests/test_scenario_engine.py
"""Tests for What-If Scenario Engine"""
from scenario_engine import generate_scenarios
from scorer import COMPONENT_WEIGHTS
class TestGenerateScenarios:
"""Test scenario generation."""
def test_returns_4_scenarios(self):
scores = {k: 50 for k in COMPONENT_WEIGHTS}
scenarios = generate_scenarios(scores)
assert len(scenarios) == 4
def test_scenario_names(self):
scores = {k: 50 for k in COMPONENT_WEIGHTS}
scenarios = generate_scenarios(scores)
names = {s["name"] for s in scenarios}
assert "Breadth Deterioration" in names
assert "Distribution Reset" in names
assert "Full Deterioration" in names
assert "Recovery" in names
def test_breadth_deterioration_raises_score(self):
"""Low breadth -> raising to 55 should increase composite."""
scores = {k: 30 for k in COMPONENT_WEIGHTS}
scores["breadth_divergence"] = 10
scenarios = generate_scenarios(scores)
breadth_scenario = next(s for s in scenarios if s["name"] == "Breadth Deterioration")
assert breadth_scenario["delta"] >= 0
def test_recovery_lowers_score(self):
"""Reducing top 2 strongest by -30pt should lower composite."""
scores = {
"distribution_days": 80,
"leading_stocks": 70,
"defensive_rotation": 60,
"breadth_divergence": 50,
"index_technical": 40,
"sentiment": 30,
}
scenarios = generate_scenarios(scores)
recovery = next(s for s in scenarios if s["name"] == "Recovery")
assert recovery["delta"] <= 0
def test_distribution_reset_lowers_high_score(self):
"""Distribution Reset should reduce score when distribution_days > 55."""
scores = {
"distribution_days": 100,
"leading_stocks": 50,
"defensive_rotation": 50,
"breadth_divergence": 50,
"index_technical": 50,
"sentiment": 50,
}
scenarios = generate_scenarios(scores)
reset = next(s for s in scenarios if s["name"] == "Distribution Reset")
assert reset["delta"] < 0, (
f"Distribution Reset with dd=100 should lower score, got delta={reset['delta']}"
)
assert reset["changes"]["distribution_days"] == 55
def test_distribution_reset_no_change_when_below_55(self):
"""Distribution Reset should not raise score when distribution_days < 55."""
scores = {
"distribution_days": 30,
"leading_stocks": 50,
"defensive_rotation": 50,
"breadth_divergence": 50,
"index_technical": 50,
"sentiment": 50,
}
scenarios = generate_scenarios(scores)
reset = next(s for s in scenarios if s["name"] == "Distribution Reset")
assert reset["delta"] == 0
assert reset["changes"]["distribution_days"] == 30
def test_scores_clamped_0_100(self):
"""Scores should never exceed 0-100 range."""
scores = {k: 90 for k in COMPONENT_WEIGHTS}
scenarios = generate_scenarios(scores)
for s in scenarios:
assert 0 <= s["new_score"] <= 100
def test_deterministic_with_equal_scores(self):
"""Equal scores should still produce deterministic results."""
scores = {k: 50 for k in COMPONENT_WEIGHTS}
s1 = generate_scenarios(scores)
s2 = generate_scenarios(scores)
for a, b in zip(s1, s2):
assert a["new_score"] == b["new_score"]
scripts/tests/test_scorer.py
"""Tests for Composite Scorer and FTD Detection"""
from scorer import (
COMPONENT_WEIGHTS,
CORRELATION_DISCOUNT,
_apply_correlation_adjustment,
calculate_composite_score,
detect_follow_through_day,
)
class TestCompositeScore:
"""Test composite scoring."""
def test_all_zeros(self):
scores = {k: 0 for k in COMPONENT_WEIGHTS}
result = calculate_composite_score(scores)
assert result["composite_score"] == 0
assert "Green" in result["zone"]
def test_all_100(self):
"""All 100 with correlation discount: 100 - 0.15*(100-80) = 97.0."""
scores = {k: 100 for k in COMPONENT_WEIGHTS}
result = calculate_composite_score(scores)
assert result["composite_score"] == 97.0
assert "Critical" in result["zone"]
def test_moderate_risk(self):
"""Calibration target: ~48-50 for moderate risk."""
scores = {
"distribution_days": 45,
"leading_stocks": 52,
"defensive_rotation": 82,
"breadth_divergence": 20,
"index_technical": 42,
"sentiment": 62,
}
result = calculate_composite_score(scores)
assert 40 <= result["composite_score"] <= 55
def test_weights_sum_to_1(self):
total = sum(COMPONENT_WEIGHTS.values())
assert abs(total - 1.0) < 0.001
def test_strongest_weakest(self):
scores = {
"distribution_days": 90,
"leading_stocks": 10,
"defensive_rotation": 50,
"breadth_divergence": 50,
"index_technical": 50,
"sentiment": 50,
}
result = calculate_composite_score(scores)
assert result["strongest_warning"]["component"] == "distribution_days"
assert result["weakest_warning"]["component"] == "leading_stocks"
def test_strongest_weakest_excludes_unavailable(self):
"""Unavailable components must not appear as strongest/weakest."""
scores = {
"distribution_days": 30,
"leading_stocks": 40,
"defensive_rotation": 50,
"breadth_divergence": 95, # Highest but unavailable
"index_technical": 5, # Lowest but unavailable
"sentiment": 60,
}
avail = {k: True for k in COMPONENT_WEIGHTS}
avail["breadth_divergence"] = False
avail["index_technical"] = False
result = calculate_composite_score(scores, avail)
# strongest should be sentiment (60), not breadth (95)
assert result["strongest_warning"]["component"] == "sentiment"
# weakest should be distribution_days (30), not index_technical (5)
assert result["weakest_warning"]["component"] == "distribution_days"
def test_contribution_sum_matches_composite(self):
"""Sum of weighted contributions should approximate composite score."""
scores = {
"distribution_days": 45,
"leading_stocks": 52,
"defensive_rotation": 30,
"breadth_divergence": 20,
"index_technical": 42,
"sentiment": 62,
}
avail = {k: True for k in COMPONENT_WEIGHTS}
avail["sentiment"] = False
result = calculate_composite_score(scores, avail)
contributions = [
v["weighted_contribution"]
for k, v in result["component_scores"].items()
if avail.get(k, True)
]
assert abs(sum(contributions) - result["composite_score"]) <= 0.5
def test_single_available_component(self):
"""Only 1 available component should still produce valid strongest/weakest."""
scores = {k: 50 for k in COMPONENT_WEIGHTS}
scores["distribution_days"] = 70
avail = {k: False for k in COMPONENT_WEIGHTS}
avail["distribution_days"] = True
result = calculate_composite_score(scores, avail)
assert result["strongest_warning"]["component"] == "distribution_days"
assert result["weakest_warning"]["component"] == "distribution_days"
assert result["composite_score"] == 70.0
class TestDataQuality:
"""Test data quality tracking."""
def test_all_available(self):
scores = {k: 50 for k in COMPONENT_WEIGHTS}
avail = {k: True for k in COMPONENT_WEIGHTS}
result = calculate_composite_score(scores, avail)
dq = result["data_quality"]
assert dq["available_count"] == 6
assert "Complete" in dq["label"]
assert dq["missing_components"] == []
def test_some_missing(self):
scores = {k: 50 for k in COMPONENT_WEIGHTS}
avail = {k: True for k in COMPONENT_WEIGHTS}
avail["breadth_divergence"] = False
avail["sentiment"] = False
result = calculate_composite_score(scores, avail)
dq = result["data_quality"]
assert dq["available_count"] == 4
assert "Partial" in dq["label"]
assert len(dq["missing_components"]) == 2
def test_many_missing(self):
scores = {k: 50 for k in COMPONENT_WEIGHTS}
avail = {k: False for k in COMPONENT_WEIGHTS}
avail["distribution_days"] = True
avail["leading_stocks"] = True
result = calculate_composite_score(scores, avail)
dq = result["data_quality"]
assert dq["available_count"] == 2
assert "Limited" in dq["label"]
def test_backward_compat_no_availability(self):
"""Without data_availability, all assumed available."""
scores = {k: 50 for k in COMPONENT_WEIGHTS}
result = calculate_composite_score(scores)
dq = result["data_quality"]
assert dq["available_count"] == 6
assert "Complete" in dq["label"]
def test_weight_redistribution_when_missing(self):
"""Missing components' weight should be redistributed, not use neutral 50."""
scores_all_80 = {k: 80 for k in COMPONENT_WEIGHTS}
# All available: composite ~ 80 (minus correlation discount)
result_full = calculate_composite_score(scores_all_80)
# Mark sentiment as unavailable: should still be ~80 from remaining
avail = {k: True for k in COMPONENT_WEIGHTS}
avail["sentiment"] = False
result_partial = calculate_composite_score(scores_all_80, avail)
# With redistribution, score should be similar (both driven by 80s)
assert abs(result_full["composite_score"] - result_partial["composite_score"]) <= 3.0
def test_weight_redistribution_excludes_unavailable(self):
"""Unavailable component with score 0 should not drag composite down."""
scores = {k: 80 for k in COMPONENT_WEIGHTS}
scores["breadth_divergence"] = 0 # Would drag down if included
avail = {k: True for k in COMPONENT_WEIGHTS}
avail["breadth_divergence"] = False # But it's unavailable
result = calculate_composite_score(scores, avail)
# Without breadth (score=0), composite should be purely from 80-scored components
# All remaining are 80, so composite ~ 80 (minus correlation discount)
assert result["composite_score"] >= 75
class TestCorrelationAdjustment:
"""Test correlation adjustment between distribution_days and defensive_rotation."""
def test_both_extreme_applies_discount(self):
"""Both >= 80 -> defensive_rotation discounted by 0.8x."""
scores = {k: 50 for k in COMPONENT_WEIGHTS}
scores["distribution_days"] = 90
scores["defensive_rotation"] = 85
result = _apply_correlation_adjustment(scores)
adj = result["adjusted_scores"]
assert adj["defensive_rotation"] == 85 * CORRELATION_DISCOUNT
assert adj["distribution_days"] == 90 # Higher weight, not discounted
assert len(result["adjustments"]) == 1
def test_one_below_threshold_no_discount(self):
"""One below threshold -> no adjustment."""
scores = {k: 50 for k in COMPONENT_WEIGHTS}
scores["distribution_days"] = 90
scores["defensive_rotation"] = 70 # Below 80
result = _apply_correlation_adjustment(scores)
assert result["adjusted_scores"]["defensive_rotation"] == 70
assert len(result["adjustments"]) == 0
def test_zone_unchanged_for_moderate(self):
"""Moderate scores should not trigger correlation adjustment."""
scores = {k: 40 for k in COMPONENT_WEIGHTS}
result = _apply_correlation_adjustment(scores)
assert len(result["adjustments"]) == 0
assert result["adjusted_scores"] == scores
def test_adjustment_details_present(self):
"""Adjustment details should include pair, original, adjusted, discount."""
scores = {k: 80 for k in COMPONENT_WEIGHTS}
result = _apply_correlation_adjustment(scores)
assert len(result["adjustments"]) == 1
adj = result["adjustments"][0]
assert "distribution_days" in adj["pair"]
assert "defensive_rotation" in adj["pair"]
assert adj["discounted_component"] == "defensive_rotation"
assert adj["original_score"] == 80
assert adj["adjusted_score"] == 80 * CORRELATION_DISCOUNT
assert adj["discount_factor"] == CORRELATION_DISCOUNT
class TestFollowThroughDay:
"""Test O'Neil-strict FTD detection."""
def test_not_applicable_below_40(self):
result = detect_follow_through_day([], 30.0)
assert result["applicable"] is False
def test_insufficient_data(self):
result = detect_follow_through_day([{"close": 100}] * 5, 50.0)
assert result["ftd_detected"] is False
def test_no_swing_low(self):
"""Flat market → no swing low found."""
history = [{"close": 100, "volume": 1000000, "date": f"day-{i}"} for i in range(30)]
result = detect_follow_through_day(history, 50.0)
assert result["ftd_detected"] is False
def test_ftd_detected_on_clear_pattern(self):
"""
Create a clear pattern:
- Days 0-9: prices rise from 100 to 109 (establishing high)
- Days 10-14: decline from 109 to 100 (5+ down days, ~8% drop)
- Day 15: swing low at 100
- Day 16: rally day 1 (up from 100 to 101)
- Day 17-18: rally continues (101.5, 102)
- Day 19 (rally day 4): FTD - big gain +2% on higher volume
History is most-recent-first, so reversed.
"""
bars = []
# Build chronologically, then reverse
# Days 0-9: uptrend
for i in range(10):
bars.append({"close": 100 + i, "volume": 1000000, "date": f"2026-01-{i + 1:02d}"})
# Days 10-14: decline (5 down days from 109)
decline_prices = [107, 105, 103, 101, 100]
for i, p in enumerate(decline_prices):
bars.append({"close": p, "volume": 1100000, "date": f"2026-01-{11 + i:02d}"})
# Day 15: swing low
bars.append({"close": 99.5, "volume": 1200000, "date": "2026-01-16"})
# Day 16: rally day 1
bars.append({"close": 101, "volume": 1000000, "date": "2026-01-17"})
# Day 17-18: rally continues
bars.append({"close": 101.5, "volume": 1000000, "date": "2026-01-18"})
bars.append({"close": 102, "volume": 1000000, "date": "2026-01-19"})
# Day 19: FTD - +2% gain on higher volume
bars.append({"close": 104.04, "volume": 1500000, "date": "2026-01-20"})
# Reverse to most-recent-first
history = list(reversed(bars))
result = detect_follow_through_day(history, 55.0)
assert result["ftd_detected"] is True
assert result["rally_day_count"] >= 4
def test_rally_reset_below_swing_low(self):
"""Rally resets if price drops below swing low."""
bars = []
# Uptrend
for i in range(10):
bars.append({"close": 100 + i, "volume": 1000000, "date": f"d-{i}"})
# Decline
for i, p in enumerate([107, 105, 103, 101, 100]):
bars.append({"close": p, "volume": 1100000, "date": f"d-{10 + i}"})
# Swing low
bars.append({"close": 99, "volume": 1200000, "date": "d-15"})
# Brief rally
bars.append({"close": 100, "volume": 1000000, "date": "d-16"})
# Drop below swing low → reset
bars.append({"close": 98, "volume": 1000000, "date": "d-17"})
# New rally from lower base
bars.append({"close": 99, "volume": 1000000, "date": "d-18"})
bars.append({"close": 99.5, "volume": 1000000, "date": "d-19"})
history = list(reversed(bars))
result = detect_follow_through_day(history, 55.0)
# FTD should NOT be detected (rally reset, new rally too short)
assert result["ftd_detected"] is False
def test_ftd_reset_requires_up_day_for_rally_day_1(self):
"""After reset (new low), flat days should NOT count as Rally Day 1.
Uses a long rally (21+ days) between swing low and crash so that
the crash bar's 20-day lookback has < 3 down days, preventing the
swing low finder from recognizing the crash as a new swing low.
"""
bars = []
# Days 0-5: uptrend
for i in range(6):
bars.append({"close": 100 + i * 2, "volume": 1000000, "date": f"d-{i}"})
# Days 6-10: decline (5 down days)
for i, p in enumerate([108, 106, 104, 100, 97]):
bars.append({"close": p, "volume": 1100000, "date": f"d-{6 + i}"})
# Day 11: swing low at 95
bars.append({"close": 95, "volume": 1200000, "date": "d-11"})
# Days 12-32: rally (21 bars, climbing from 96 to 116)
for i in range(21):
bars.append({"close": 96 + i, "volume": 1000000, "date": f"d-{12 + i}"})
# Day 33: crash below 95 (only 1 down day in 20-bar lookback)
bars.append({"close": 94, "volume": 1000000, "date": "d-33"})
# Day 34: flat at 94 — should NOT be Rally Day 1
bars.append({"close": 94, "volume": 1000000, "date": "d-34"})
# Day 35: down — NOT Rally Day 1
bars.append({"close": 93.5, "volume": 1000000, "date": "d-35"})
# Day 36: up from 93.5 → THIS should be the new Rally Day 1
bars.append({"close": 95, "volume": 1000000, "date": "d-36"})
# Days 37-38
bars.append({"close": 96, "volume": 1000000, "date": "d-37"})
bars.append({"close": 97, "volume": 1000000, "date": "d-38"})
history = list(reversed(bars))
result = detect_follow_through_day(history, 55.0)
# After reset, flat/down days skipped, Day 1 at d-36, days 2-3
assert result["rally_day_count"] == 3
assert result["reset_count"] >= 1
def test_ftd_reset_clears_invalidation(self):
"""After reset, finding new Rally Day 1 leads to valid FTD detection."""
bars = []
# Days 0-5: uptrend
for i in range(6):
bars.append({"close": 100 + i * 2, "volume": 1000000, "date": f"d-{i}"})
# Days 6-10: decline (5 down days)
for i, p in enumerate([108, 106, 104, 100, 97]):
bars.append({"close": p, "volume": 1100000, "date": f"d-{6 + i}"})
# Day 11: swing low at 95
bars.append({"close": 95, "volume": 1200000, "date": "d-11"})
# Days 12-32: rally (21 bars, climbing from 96 to 116)
for i in range(21):
bars.append({"close": 96 + i, "volume": 1000000, "date": f"d-{12 + i}"})
# Day 33: crash below 95
bars.append({"close": 94, "volume": 1000000, "date": "d-33"})
# Day 34: new Rally Day 1 (up from 94)
bars.append({"close": 96, "volume": 1000000, "date": "d-34"})
# Days 35-36
bars.append({"close": 97, "volume": 1000000, "date": "d-35"})
bars.append({"close": 98, "volume": 1000000, "date": "d-36"})
# Day 37 (rally day 4): FTD - +2% gain on higher volume
bars.append({"close": 99.96, "volume": 1500000, "date": "d-37"})
history = list(reversed(bars))
result = detect_follow_through_day(history, 55.0)
assert result["ftd_detected"] is True
assert result["rally_day_count"] >= 4
assert result["reset_count"] >= 1
def test_ftd_day1_accepts_range_top_half(self):
"""Rally Day 1 can be a bar closing in upper 50% of range, not just up day.
Constructs a scenario where the bar after the swing low has
close == prev_close (not an up day) but closes in the upper half
of its intraday range.
"""
bars = []
# Uptrend
for i in range(10):
bars.append({"close": 100 + i, "volume": 1000000, "date": f"d-{i}"})
# Decline
for i, p in enumerate([107, 105, 103, 101, 100]):
bars.append({"close": p, "volume": 1100000, "date": f"d-{10 + i}"})
# Swing low at 95
bars.append({"close": 95, "volume": 1200000, "date": "d-15"})
# Next bar: close=95 (same as swing low, NOT an up day), but upper half range
# high=98, low=92, close=95 → range_position = (95-92)/(98-92) = 0.5
bars.append({"close": 95, "high": 98, "low": 92, "volume": 1000000, "date": "d-16"})
# Day 2, 3
bars.append({"close": 97, "volume": 1000000, "date": "d-17"})
bars.append({"close": 98, "volume": 1000000, "date": "d-18"})
history = list(reversed(bars))
result = detect_follow_through_day(history, 55.0)
# The finder may find d-16 or d-15 as swing low (both close=95).
# If d-16 is the swing low, Day 1 would be d-17 (up from 95).
# If d-15 is the swing low, d-16 qualifies via range_top_half.
# Either way, we should have a rally of at least 2 days.
assert result["rally_day_count"] >= 2
scripts/tests/test_sentiment.py
"""Tests for Sentiment Calculator"""
from calculators.sentiment_calculator import calculate_sentiment
class TestSentimentMissingData:
"""Test missing data handling."""
def test_all_none_returns_50(self):
"""All inputs None -> score=50, data_available=False."""
result = calculate_sentiment()
assert result["score"] == 50
assert result["data_available"] is False
assert "NO DATA" in result["signal"]
def test_partial_data_is_available(self):
"""At least one input -> data_available=True."""
result = calculate_sentiment(vix_level=15.0)
assert result["data_available"] is True
def test_vix_only(self):
result = calculate_sentiment(vix_level=15.0)
assert result["score"] == 8 # VIX 15 -> +8pt
def test_put_call_only(self):
result = calculate_sentiment(put_call_ratio=0.65)
assert result["score"] == 25 # PC < 0.70 -> +25pt
class TestSentimentScoring:
"""Boundary tests for sentiment scoring."""
def test_extreme_complacency_without_margin(self):
"""Low VIX + low P/C + steep contango (no margin) = 85."""
result = calculate_sentiment(
vix_level=11.0,
put_call_ratio=0.55,
vix_term_structure="steep_contango",
)
# 25+35+25=85
assert result["score"] == 85
def test_extreme_complacency_with_margin(self):
"""All components maxed = 100."""
result = calculate_sentiment(
vix_level=11.0,
put_call_ratio=0.55,
vix_term_structure="steep_contango",
margin_debt_yoy_pct=35.0,
)
# 25+35+25+15=100
assert result["score"] == 100
def test_high_fear(self):
"""High VIX + high P/C + backwardation = 0."""
result = calculate_sentiment(
vix_level=30.0,
put_call_ratio=0.95,
vix_term_structure="backwardation",
)
# -8+0+(-8) = -16 -> clamped to 0
assert result["score"] == 0
def test_moderate_sentiment(self):
result = calculate_sentiment(
vix_level=15.0,
put_call_ratio=0.75,
vix_term_structure="contango",
)
# VIX 15->8, PC 0.75->12, contango->12 = 32
assert result["score"] == 32
def test_margin_debt_scored(self):
"""Margin debt should now affect score."""
without = calculate_sentiment(vix_level=15.0)
with_margin = calculate_sentiment(vix_level=15.0, margin_debt_yoy_pct=40.0)
assert with_margin["score"] > without["score"]
def test_margin_debt_thresholds(self):
"""Test margin debt scoring thresholds."""
# <= 10% -> 0pt
r1 = calculate_sentiment(vix_level=15.0, margin_debt_yoy_pct=5.0)
assert r1["score"] == 8 # VIX only
# > 10% -> 5pt
r2 = calculate_sentiment(vix_level=15.0, margin_debt_yoy_pct=15.0)
assert r2["score"] == 13 # 8 + 5
# > 20% -> 10pt
r3 = calculate_sentiment(vix_level=15.0, margin_debt_yoy_pct=25.0)
assert r3["score"] == 18 # 8 + 10
# > 30% -> 15pt
r4 = calculate_sentiment(vix_level=15.0, margin_debt_yoy_pct=36.0)
assert r4["score"] == 23 # 8 + 15
def test_margin_debt_details_has_points(self):
"""Margin debt details should include points field."""
result = calculate_sentiment(vix_level=15.0, margin_debt_yoy_pct=25.0)
md = result["details"]["margin_debt"]
assert "points" in md
assert md["points"] == 10
assert "note" not in md # Old "not scored" note removed
scripts/tests/test_utils.py
"""Tests for shared utilities."""
from datetime import date
from utils import count_business_days
class TestCountBusinessDays:
"""Test business day counting."""
def test_same_day(self):
"""Same day -> 0 business days."""
d = date(2026, 3, 16) # Monday
assert count_business_days(d, d) == 0
def test_fri_to_mon(self):
"""Friday to Monday = 1 business day."""
fri = date(2026, 3, 13) # Friday
mon = date(2026, 3, 16) # Monday
assert count_business_days(fri, mon) == 1
def test_fri_to_wed(self):
"""Friday to Wednesday = 3 business days."""
fri = date(2026, 3, 13) # Friday
wed = date(2026, 3, 18) # Wednesday
assert count_business_days(fri, wed) == 3
def test_mon_to_fri(self):
"""Monday to Friday = 4 business days."""
mon = date(2026, 3, 16) # Monday
fri = date(2026, 3, 20) # Friday
assert count_business_days(mon, fri) == 4
def test_future_returns_negative_one(self):
"""start > end -> returns -1."""
future = date(2026, 3, 20)
past = date(2026, 3, 16)
assert count_business_days(future, past) == -1
def test_full_week(self):
"""Monday to next Monday = 5 business days."""
mon1 = date(2026, 3, 16) # Monday
mon2 = date(2026, 3, 23) # Next Monday
assert count_business_days(mon1, mon2) == 5
def test_weekend_only(self):
"""Saturday to Sunday = 0 business days."""
sat = date(2026, 3, 14) # Saturday
sun = date(2026, 3, 15) # Sunday
assert count_business_days(sat, sun) == 0
scripts/utils.py
"""
Market Top Detector - Shared Utilities
Common helper functions used across multiple modules.
"""
from datetime import date
from _market_calendar import count_sessions
def count_business_days(start_date: date, end_date: date) -> int:
"""Count business days between start (exclusive) and end (inclusive).
Friday→Monday = 1 business day.
Returns -1 if start_date > end_date (future date).
"""
if start_date > end_date:
return -1
return count_sessions(
"XNYS",
start_date,
end_date,
include_start=False,
include_end=True,
)
SKILL.md
---
name: market-top-detector
description: Assesses heuristic market top risk using O'Neil Distribution Days, Minervini Leading Stock Deterioration, and Monty Defensive Sector Rotation. Generates a 0-100 composite score with risk zone classification. Use when user asks about market top risk, distribution days, defensive rotation, leadership breakdown, or whether to reduce equity exposure. Focuses on 2-8 week tactical timing signals for 10-20% corrections.
---
# Market Top Detector Skill
## Purpose
Assess heuristic market top risk using a 6-component scoring system (0-100). Integrates three market risk methodologies:
1. **O'Neil** - Distribution Day accumulation (institutional selling)
2. **Minervini** - Leading stock deterioration pattern
3. **Monty** - Defensive sector rotation signal
Unlike the Bubble Detector (macro/multi-month evaluation), this skill focuses on **tactical 2-8 week timing signals** that precede 10-20% market corrections.
## Score interpretation
Treat the 0-100 composite as a **heuristic risk score**, not an estimated event
probability. A score of 70 does not mean a 70% chance of a top. Risk-budget
percentages are policy heuristics, not measured outcome frequencies.
Historical patterns are retrospective estimates, not point-in-time evidence.
Report **no comparable episode** when any component is unavailable/invalid or
nearest-pattern weighted RMS exceeds 15 points. This tolerance is a predeclared
heuristic, not an empirically validated similarity threshold.
Follow [the point-in-time evaluation protocol](references/point_in_time_evaluation.md)
before making calibration claims. No measured calibration results are supplied.
## When to Use This Skill
**English:**
- User asks "Is the market topping?" or "Are we near a top?"
- User notices distribution days accumulating
- User observes defensive sectors outperforming growth
- User sees leading stocks breaking down while indices hold
- User asks about reducing equity exposure timing
- User wants to assess correction risk for the next 2-8 weeks
**Japanese:**
- 「天井が近い？」「今は利確すべき？」
- ディストリビューションデーの蓄積を懸念
- ディフェンシブセクターがグロースをアウトパフォーム
- 先導株が崩れ始めているが指数はまだ持ちこたえている
- エクスポージャー縮小のタイミング判断
- 今後2〜8週間の調整リスクを評価したい
## Prerequisites
**Required:**
- **FMP API Key:** Set `$FMP_API_KEY` environment variable or pass `--api-key`. Free tier sufficient (~33 API calls per execution).
- **WebSearch Access:** Required to collect S&P 500 breadth (50DMA %) and CBOE Put/Call ratio data.
**Optional:**
- **Margin Debt Data:** Enhances sentiment scoring but typically 1-2 months lagged.
- **VIX Term Structure:** Auto-detected from FMP API if VIX3M quote available; manual override via `--vix-term`.
**Data Freshness:** All manually collected data should be from the most recent 3 business days for accurate analysis.
## Difference from Bubble Detector
| Aspect | Market Top Detector | Bubble Detector |
|--------|-------------------|-----------------|
| Timeframe | 2-8 weeks | Months to years |
| Target | 10-20% correction | Bubble collapse (30%+) |
| Methodology | O'Neil/Minervini/Monty | Minsky/Kindleberger |
| Data | Price/Volume + Breadth | Valuation + Sentiment + Social |
| Score Range | 0-100 composite | 0-15 points |
---
## Execution Workflow
### Phase 1: Data Collection via WebSearch
Before running the Python script, collect the following data using WebSearch.
**Data Freshness Requirement:** All data must be from the most recent 3 business days. Stale data degrades analysis quality.
```
1. S&P 500 Breadth (200DMA above %)
AUTO-FETCHED from TraderMonty CSV (no WebSearch needed)
The script fetches this automatically from GitHub Pages CSV data.
Override: --breadth-200dma [VALUE] to use a manual value instead.
Disable: --no-auto-breadth to skip auto-fetch entirely.
2. [REQUIRED] S&P 500 Breadth (50DMA above %)
Valid range: 20-100
Primary search: "S&P 500 percent stocks above 50 day moving average"
Fallback: "market breadth 50dma site:barchart.com"
Direct fallback when search snippets are poor: fetch `https://www.barchart.com/stocks/quotes/$S5FI/overview` and extract the embedded `lastPrice` / `tradeTime` for “S&P 500 Stocks Above 50-Day Average”.
Record the data date
3. [REQUIRED] CBOE Equity Put/Call Ratio
Valid range: 0.30-1.50
Primary search: "CBOE equity put call ratio today"
Fallback: "CBOE total put call ratio current"
Fallback: "put call ratio site:cboe.com"
Direct fallback when Cboe CSV endpoints are stale: fetch `https://ycharts.com/indicators/cboe_equity_put_call_ratio` and parse the “Last Value” / “Latest Period” table fields. Treat this as a secondary source and cite it in freshness notes.
Record the data date
4. [OPTIONAL] VIX Term Structure
Values: steep_contango / contango / flat / backwardation
Primary search: "VIX VIX3M ratio term structure today"
Fallback: "VIX futures term structure contango backwardation"
Note: Auto-detected from FMP API if VIX3M quote available.
CLI --vix-term overrides auto-detection.
5. [OPTIONAL] Margin Debt YoY %
Primary search: "FINRA margin debt latest year over year percent"
Fallback: "NYSE margin debt monthly"
Note: Typically 1-2 months lagged. Record the reporting month.
```
### Phase 2: Execute Python Script
Run the script with collected data as CLI arguments:
```bash
python3 skills/market-top-detector/scripts/market_top_detector.py \
--api-key $FMP_API_KEY \
--breadth-50dma [VALUE] --breadth-50dma-date [YYYY-MM-DD] \
--put-call [VALUE] --put-call-date [YYYY-MM-DD] \
--vix-term [steep_contango|contango|flat|backwardation] \
--margin-debt-yoy [VALUE] --margin-debt-date [YYYY-MM-DD] \
--output-dir reports/ \
--context "Consumer Confidence=[VALUE]" "Gold Price=[VALUE]"
# 200DMA breadth is auto-fetched from TraderMonty CSV.
# Override with --breadth-200dma [VALUE] if needed.
# Disable with --no-auto-breadth to skip auto-fetch.
```
The script will:
1. Fetch S&P 500, QQQ, VIX quotes and history from FMP API
2. Fetch Leading ETF (ARKK, WCLD, IGV, XBI, SOXX, SMH, KWEB, TAN) data
3. Fetch Sector ETF (XLU, XLP, XLV, VNQ, XLK, XLC, XLY) data
4. Calculate all 6 components
5. Generate composite score and reports
### Phase 3: Present Results
Present the generated Markdown report to the user, highlighting:
- Composite score and risk zone
- Data freshness warnings (if any data older than 3 days)
- Strongest warning signal (highest component score)
- Historical comparison (closest past top pattern)
- What-if scenarios (sensitivity to key changes)
- Recommended actions based on risk zone
- Follow-Through Day status (if applicable)
- Delta vs previous run (if prior report exists)
---
## 6-Component Scoring System
| # | Component | Weight | Data Source | Key Signal |
|---|-----------|--------|-------------|------------|
| 1 | Distribution Day Count | **25%** | FMP API | Institutional selling in last 25 trading days |
| 2 | Leading Stock Health | **20%** | FMP API | Growth ETF basket deterioration |
| 3 | Defensive Sector Rotation | **15%** | FMP API | Defensive vs Growth relative performance |
| 4 | Market Breadth Divergence | **15%** | Auto (CSV) + WebSearch | 200DMA (auto) / 50DMA (WebSearch) breadth vs index level |
| 5 | Index Technical Condition | **15%** | FMP API | MA structure, failed rallies, lower highs |
| 6 | Sentiment & Speculation | **10%** | FMP + WebSearch | VIX, Put/Call, term structure |
## Risk Zone Mapping
| Score | Zone | Risk Budget | Action |
|-------|------|-------------|--------|
| 0-20 | Green (Normal) | 100% | Normal operations |
| 21-40 | Yellow (Early Warning) | 80-90% | Tighten stops, reduce new entries |
| 41-60 | Orange (Elevated Risk) | 60-75% | Profit-taking on weak positions |
| 61-80 | Red (High Risk) | 40-55% | Aggressive profit-taking |
| 81-100 | Critical (Very High Risk) | 20-35% | Maximum defense, hedging |
---
## Exchange Calendar and Replay
Install `requirements.txt` before running the detector. Freshness uses XNYS
sessions rather than weekdays. `--as-of YYYY-MM-DD` is accepted for the live
evaluation date, but historical live replay fails closed because the current
quote endpoints are not point-in-time sources.
## API Requirements
**Required:** FMP API key (free tier sufficient: ~33 calls per execution)
**Optional:** WebSearch data for breadth and sentiment (improves accuracy)
## Output Files
- JSON: `market_top_YYYY-MM-DD_HHMMSS.json`
- Markdown: `market_top_YYYY-MM-DD_HHMMSS.md`
## Reference Documents
### `references/market_top_methodology.md`
- Full methodology with O'Neil, Minervini, and Monty frameworks
- Component scoring details and thresholds
- Historical validation notes
### `references/distribution_day_guide.md`
- Detailed O'Neil Distribution Day rules
- Stalling day identification
- Follow-Through Day (FTD) mechanics
### `references/historical_tops.md`
- Analysis of 2000, 2007, 2018, 2022 market tops
- Component score patterns during historical tops
- Lessons learned and retrospective illustrative estimates
### When to Load References
- **First use:** Load `market_top_methodology.md` for full framework understanding
- **Distribution day questions:** Load `distribution_day_guide.md`
- **Historical context:** Load `historical_tops.md`
- **Regular execution:** References not needed - script handles scoring