Downtrend Duration Analyzer

#!/usr/bin/env python3
"""
Downtrend Duration Analyzer

Analyzes historical price data to identify downtrend periods (peak-to-trough)
and computes duration statistics segmented by sector and market cap.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd
import requests

# Market cap tier thresholds (in billions USD)
MARKET_CAP_TIERS = {
    "Mega": 200_000_000_000,  # >= $200B
    "Large": 10_000_000_000,  # $10B - $200B
    "Mid": 2_000_000_000,  # $2B - $10B
    "Small": 0,  # < $2B
}

# Default sectors for analysis
DEFAULT_SECTORS = [
    "Technology",
    "Healthcare",
    "Financial Services",
    "Consumer Cyclical",
    "Consumer Defensive",
    "Industrials",
    "Energy",
    "Basic Materials",
    "Utilities",
    "Real Estate",
    "Communication Services",
]


def get_api_key(api_key_arg: str | None) -> str:
    """Get FMP API key from argument or environment variable."""
    if api_key_arg:
        return api_key_arg
    api_key = os.environ.get("FMP_API_KEY")
    if not api_key:
        print(
            "Error: FMP API key required. Set FMP_API_KEY environment variable or use --api-key",
            file=sys.stderr,
        )
        sys.exit(1)
    return api_key


def get_market_cap_tier(market_cap: float | None) -> str:
    """Classify market cap into tier."""
    if market_cap is None:
        return "Unknown"
    if market_cap >= MARKET_CAP_TIERS["Mega"]:
        return "Mega"
    elif market_cap >= MARKET_CAP_TIERS["Large"]:
        return "Large"
    elif market_cap >= MARKET_CAP_TIERS["Mid"]:
        return "Mid"
    else:
        return "Small"


def fetch_stock_list(api_key: str, sector: str | None = None) -> list[dict]:
    """Fetch list of stocks, optionally filtered by sector.

    Uses the /stable/company-screener endpoint (the v3 /stock-screener it
    replaced 403s for keys issued after 2025-08-31), with a v3 fallback for
    legacy keys. Both take the same params and return the same fields
    (symbol, sector, marketCap, ...).
    """
    params: dict[str, Any] = {
        "apikey": api_key,
        "isActivelyTrading": "true",
        "limit": 500,
    }
    if sector:
        params["sector"] = sector
    endpoints = [
        "https://financialmodelingprep.com/stable/company-screener",
        "https://financialmodelingprep.com/api/v3/stock-screener",
    ]
    for url in endpoints:
        try:
            response = requests.get(url, params=params, timeout=30)
        except requests.RequestException as e:
            print(f"Error fetching stock list ({url}): {e}", file=sys.stderr)
            continue
        if response.status_code != 200:
            continue
        try:
            return response.json()
        except ValueError:
            continue
    print("Error fetching stock list: all screener endpoints failed", file=sys.stderr)
    return []


# --- FMP endpoint fallback: stable (new users) -> v3 (legacy users) ---
_FMP_HIST_ENDPOINTS = [
    ("https://financialmodelingprep.com/stable/historical-price-eod/full", True),
    ("https://financialmodelingprep.com/api/v3/historical-price-full", False),
]
_endpoint_failures: dict[str, int] = {}
_BREAKER_THRESHOLD = 3


def fetch_historical_prices(
    api_key: str, symbol: str, from_date: str, to_date: str
) -> pd.DataFrame:
    """Fetch historical daily prices for a symbol."""
    for base_url, is_stable in _FMP_HIST_ENDPOINTS:
        if _endpoint_failures.get(base_url, 0) >= _BREAKER_THRESHOLD:
            continue
        if is_stable:
            url = base_url
            params = {"symbol": symbol, "from": from_date, "to": to_date, "apikey": api_key}
        else:
            url = f"{base_url}/{symbol}"
            params = {"from": from_date, "to": to_date, "apikey": api_key}
        try:
            response = requests.get(url, params=params, timeout=30)
            if response.status_code != 200:
                _endpoint_failures[base_url] = _endpoint_failures.get(base_url, 0) + 1
                continue
            data = response.json()
            historical = None
            if isinstance(data, dict) and "historical" in data:
                historical = data["historical"]
            elif isinstance(data, dict) and "historicalStockList" in data:
                for entry in data["historicalStockList"]:
                    if entry.get("symbol", "").replace("-", ".") == symbol.replace("-", "."):
                        historical = entry.get("historical", [])
                        break
            if historical is not None:
                _endpoint_failures[base_url] = 0
                df = pd.DataFrame(historical)
                if df.empty:
                    return df
                df["date"] = pd.to_datetime(df["date"])
                df = df.sort_values("date").reset_index(drop=True)
                return df[["date", "open", "high", "low", "close", "volume"]]
            _endpoint_failures[base_url] = _endpoint_failures.get(base_url, 0) + 1
        except requests.RequestException:
            _endpoint_failures[base_url] = _endpoint_failures.get(base_url, 0) + 1
            continue
    print(f"Error fetching prices for {symbol}: all endpoints failed", file=sys.stderr)
    return pd.DataFrame()


def detect_peaks_troughs(
    prices: pd.DataFrame, peak_window: int = 20, trough_window: int = 20
) -> tuple[list[int], list[int]]:
    """
    Detect local peaks and troughs using rolling window.

    Returns indices of peaks and troughs in the price dataframe.
    """
    closes = prices["close"].values
    n = len(closes)
    peaks = []
    troughs = []

    for i in range(peak_window, n - peak_window):
        window_start = i - peak_window
        window_end = i + peak_window + 1
        window = closes[window_start:window_end]

        # Peak: highest in window
        if closes[i] == np.max(window):
            peaks.append(i)

        # Trough: lowest in window
        if closes[i] == np.min(window):
            troughs.append(i)

    return peaks, troughs


def find_downtrends(
    prices: pd.DataFrame,
    peaks: list[int],
    troughs: list[int],
    min_depth_pct: float = 5.0,
    min_duration_days: int = 3,
) -> list[dict]:
    """
    Identify downtrend periods from peaks to subsequent troughs.

    Returns list of downtrend dictionaries with duration and depth.
    """
    downtrends = []
    closes = prices["close"].values
    dates = prices["date"].values

    for peak_idx in peaks:
        peak_price = closes[peak_idx]
        peak_date = dates[peak_idx]

        # Find the next trough after this peak
        subsequent_troughs = [t for t in troughs if t > peak_idx]
        if not subsequent_troughs:
            continue

        # Find the lowest trough before the next peak
        next_peaks = [p for p in peaks if p > peak_idx]
        end_idx = next_peaks[0] if next_peaks else len(closes)

        valid_troughs = [t for t in subsequent_troughs if t < end_idx]
        if not valid_troughs:
            continue

        # Find the deepest trough
        trough_idx = min(valid_troughs, key=lambda t: closes[t])
        trough_price = closes[trough_idx]
        trough_date = dates[trough_idx]

        # Calculate depth and duration
        depth_pct = ((trough_price - peak_price) / peak_price) * 100
        duration_days = int(trough_idx - peak_idx)

        # Apply filters
        if abs(depth_pct) < min_depth_pct:
            continue
        if duration_days < min_duration_days:
            continue

        downtrends.append(
            {
                "peak_idx": peak_idx,
                "trough_idx": trough_idx,
                "peak_date": pd.Timestamp(peak_date).strftime("%Y-%m-%d"),
                "trough_date": pd.Timestamp(trough_date).strftime("%Y-%m-%d"),
                "peak_price": float(peak_price),
                "trough_price": float(trough_price),
                "duration_days": duration_days,
                "depth_pct": round(depth_pct, 2),
            }
        )

    return downtrends


def analyze_symbol(
    api_key: str,
    symbol: str,
    sector: str,
    market_cap: float | None,
    from_date: str,
    to_date: str,
    peak_window: int,
    trough_window: int,
    min_depth_pct: float,
) -> list[dict]:
    """Analyze downtrends for a single symbol."""
    prices = fetch_historical_prices(api_key, symbol, from_date, to_date)

    if prices.empty or len(prices) < peak_window * 2 + 1:
        return []

    peaks, troughs = detect_peaks_troughs(prices, peak_window, trough_window)

    if not peaks or not troughs:
        return []

    downtrends = find_downtrends(prices, peaks, troughs, min_depth_pct)

    market_cap_tier = get_market_cap_tier(market_cap)

    # Add metadata to each downtrend
    for dt in downtrends:
        dt["symbol"] = symbol
        dt["sector"] = sector
        dt["market_cap_tier"] = market_cap_tier

    return downtrends


def compute_statistics(downtrends: list[dict]) -> dict[str, Any]:
    """Compute summary statistics from downtrend list."""
    if not downtrends:
        return {
            "total_downtrends": 0,
            "median_duration_days": 0,
            "mean_duration_days": 0,
            "p25_duration_days": 0,
            "p75_duration_days": 0,
            "p90_duration_days": 0,
        }

    durations = [dt["duration_days"] for dt in downtrends]
    return {
        "total_downtrends": len(downtrends),
        "median_duration_days": int(np.median(durations)),
        "mean_duration_days": round(np.mean(durations), 1),
        "p25_duration_days": int(np.percentile(durations, 25)),
        "p75_duration_days": int(np.percentile(durations, 75)),
        "p90_duration_days": int(np.percentile(durations, 90)),
    }


def group_statistics(downtrends: list[dict], group_key: str) -> dict[str, dict[str, Any]]:
    """Compute statistics grouped by a key (sector or market_cap_tier)."""
    groups: dict[str, list[dict]] = {}

    for dt in downtrends:
        key = dt.get(group_key, "Unknown")
        if key not in groups:
            groups[key] = []
        groups[key].append(dt)

    result = {}
    for key, group_downtrends in groups.items():
        durations = [dt["duration_days"] for dt in group_downtrends]
        result[key] = {
            "count": len(group_downtrends),
            "median_days": int(np.median(durations)),
            "mean_days": round(np.mean(durations), 1),
        }

    return result


def generate_markdown_report(analysis_result: dict[str, Any], output_path: Path) -> None:
    """Generate markdown report from analysis results."""
    params = analysis_result["parameters"]
    summary = analysis_result["summary"]
    by_sector = analysis_result.get("by_sector", {})
    by_market_cap = analysis_result.get("by_market_cap", {})

    lines = [
        "# Downtrend Duration Analysis",
        "",
        f"**Date**: {analysis_result['analysis_date'][:10]}",
        f"**Lookback**: {params['lookback_years']} years",
    ]

    if params.get("sector_filter"):
        lines.append(f"**Sector**: {params['sector_filter']}")

    lines.extend(
        [
            "",
            "## Summary Statistics",
            "",
            "| Metric | Value |",
            "|--------|-------|",
            f"| Total Downtrends | {summary['total_downtrends']:,} |",
            f"| Median Duration | {summary['median_duration_days']} days |",
            f"| Mean Duration | {summary['mean_duration_days']} days |",
            f"| 25th Percentile | {summary['p25_duration_days']} days |",
            f"| 75th Percentile | {summary['p75_duration_days']} days |",
            f"| 90th Percentile | {summary['p90_duration_days']} days |",
            "",
        ]
    )

    if by_market_cap:
        lines.extend(
            [
                "## By Market Cap Tier",
                "",
                "| Tier | Count | Median | Mean |",
                "|------|-------|--------|------|",
            ]
        )
        tier_order = ["Mega", "Large", "Mid", "Small"]
        for tier in tier_order:
            if tier in by_market_cap:
                stats = by_market_cap[tier]
                lines.append(
                    f"| {tier} | {stats['count']} | {stats['median_days']} days | {stats['mean_days']} days |"
                )
        lines.append("")

    if by_sector:
        lines.extend(
            [
                "## By Sector",
                "",
                "| Sector | Count | Median | Mean |",
                "|--------|-------|--------|------|",
            ]
        )
        for sector, stats in sorted(by_sector.items(), key=lambda x: x[1]["median_days"]):
            lines.append(
                f"| {sector} | {stats['count']} | {stats['median_days']} days | {stats['mean_days']} days |"
            )
        lines.append("")

    lines.extend(
        [
            "## Key Insights",
            "",
            "1. **Percentile Guidance**: Use P50 (median) for typical expectations; P75-P90 for conservative planning",
            "2. **Market Cap Effect**: Larger companies typically recover faster from corrections",
            "3. **Sector Variation**: Defensive sectors show shorter corrections than cyclical sectors",
            "",
        ]
    )

    output_path.write_text("\n".join(lines))


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Analyze historical downtrend durations by sector and market cap"
    )
    parser.add_argument(
        "--api-key",
        help="FMP API key (or set FMP_API_KEY env var)",
    )
    parser.add_argument(
        "--sector",
        help="Filter to specific sector (e.g., 'Technology')",
    )
    parser.add_argument(
        "--lookback-years",
        type=int,
        default=5,
        help="Years of historical data to analyze (default: 5)",
    )
    parser.add_argument(
        "--peak-window",
        type=int,
        default=20,
        help="Rolling window size for peak detection (default: 20)",
    )
    parser.add_argument(
        "--trough-window",
        type=int,
        default=20,
        help="Rolling window size for trough detection (default: 20)",
    )
    parser.add_argument(
        "--min-depth",
        type=float,
        default=5.0,
        help="Minimum depth percentage for a downtrend (default: 5.0)",
    )
    parser.add_argument(
        "--max-stocks",
        type=int,
        default=100,
        help="Maximum stocks to analyze (default: 100)",
    )
    parser.add_argument(
        "--output-dir",
        type=str,
        default="reports",
        help="Output directory for reports (default: reports)",
    )

    args = parser.parse_args()

    api_key = get_api_key(args.api_key)

    # Calculate date range
    to_date = datetime.now().strftime("%Y-%m-%d")
    from_date = (datetime.now() - timedelta(days=365 * args.lookback_years)).strftime("%Y-%m-%d")

    print(f"Analyzing downtrends from {from_date} to {to_date}")

    # Get stock list
    stocks = fetch_stock_list(api_key, args.sector)
    if not stocks:
        print("No stocks found matching criteria", file=sys.stderr)
        sys.exit(1)

    stocks = stocks[: args.max_stocks]
    print(f"Analyzing {len(stocks)} stocks...")

    # Analyze each stock
    all_downtrends: list[dict] = []
    for i, stock in enumerate(stocks):
        symbol = stock.get("symbol", "")
        sector = stock.get("sector", "Unknown")
        market_cap = stock.get("marketCap")

        if i % 10 == 0:
            print(f"  Progress: {i}/{len(stocks)} stocks processed")

        downtrends = analyze_symbol(
            api_key,
            symbol,
            sector,
            market_cap,
            from_date,
            to_date,
            args.peak_window,
            args.trough_window,
            args.min_depth,
        )
        all_downtrends.extend(downtrends)

    print(f"Found {len(all_downtrends)} downtrend periods")

    # Compute statistics
    summary = compute_statistics(all_downtrends)
    by_sector = group_statistics(all_downtrends, "sector")
    by_market_cap = group_statistics(all_downtrends, "market_cap_tier")

    # Build result
    result = {
        "schema_version": "1.0",
        "analysis_date": datetime.now().isoformat() + "Z",
        "parameters": {
            "lookback_years": args.lookback_years,
            "sector_filter": args.sector,
            "peak_window": args.peak_window,
            "trough_window": args.trough_window,
            "min_depth_pct": args.min_depth,
        },
        "summary": summary,
        "by_sector": by_sector,
        "by_market_cap": by_market_cap,
        "downtrends": all_downtrends,
    }

    # Create output directory
    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")

    # Write JSON
    json_path = output_dir / f"downtrend_analysis_{timestamp}.json"
    with open(json_path, "w") as f:
        json.dump(result, f, indent=2)
    print(f"JSON report saved to: {json_path}")

    # Write Markdown
    md_path = output_dir / f"downtrend_analysis_{timestamp}.md"
    generate_markdown_report(result, md_path)
    print(f"Markdown report saved to: {md_path}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Generate Interactive HTML Histogram for Downtrend Duration Analysis

Creates Plotly.js-based interactive visualizations from downtrend analysis JSON data.
"""

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Any

# HTML template with embedded Plotly.js
HTML_TEMPLATE = """<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Downtrend Duration Analysis</title>
    <script src="https://cdn.plot.ly/plotly-2.27.0.min.js"></script>
    <style>
        body {{
            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
            margin: 0;
            padding: 20px;
            background: #f5f5f5;
        }}
        .container {{
            max-width: 1400px;
            margin: 0 auto;
            background: white;
            padding: 20px;
            border-radius: 8px;
            box-shadow: 0 2px 4px rgba(0,0,0,0.1);
        }}
        h1 {{
            color: #333;
            margin-bottom: 10px;
        }}
        .subtitle {{
            color: #666;
            margin-bottom: 20px;
        }}
        .controls {{
            display: flex;
            gap: 20px;
            margin-bottom: 20px;
            flex-wrap: wrap;
        }}
        .control-group {{
            display: flex;
            flex-direction: column;
            gap: 5px;
        }}
        .control-group label {{
            font-weight: 600;
            color: #444;
            font-size: 14px;
        }}
        select {{
            padding: 8px 12px;
            border: 1px solid #ddd;
            border-radius: 4px;
            font-size: 14px;
            min-width: 150px;
        }}
        .stats-grid {{
            display: grid;
            grid-template-columns: repeat(auto-fit, minmax(150px, 1fr));
            gap: 15px;
            margin-bottom: 20px;
        }}
        .stat-card {{
            background: #f8f9fa;
            padding: 15px;
            border-radius: 6px;
            text-align: center;
        }}
        .stat-value {{
            font-size: 24px;
            font-weight: 700;
            color: #2563eb;
        }}
        .stat-label {{
            font-size: 12px;
            color: #666;
            margin-top: 5px;
        }}
        #histogram {{
            width: 100%;
            height: 500px;
        }}
        #boxplot {{
            width: 100%;
            height: 300px;
            margin-top: 20px;
        }}
        .footer {{
            margin-top: 20px;
            padding-top: 15px;
            border-top: 1px solid #eee;
            font-size: 12px;
            color: #888;
        }}
    </style>
</head>
<body>
    <div class="container">
        <h1>Downtrend Duration Analysis</h1>
        <div class="subtitle">
            Analysis Date: {analysis_date} | Lookback: {lookback_years} years | Total Downtrends: {total_downtrends:,}
        </div>

        <div class="controls">
            <div class="control-group">
                <label for="sectorFilter">Sector</label>
                <select id="sectorFilter" onchange="updateCharts()">
                    <option value="all">All Sectors</option>
                    {sector_options}
                </select>
            </div>
            <div class="control-group">
                <label for="marketCapFilter">Market Cap</label>
                <select id="marketCapFilter" onchange="updateCharts()">
                    <option value="all">All Market Caps</option>
                    <option value="Mega">Mega ($200B+)</option>
                    <option value="Large">Large ($10B-$200B)</option>
                    <option value="Mid">Mid ($2B-$10B)</option>
                    <option value="Small">Small (<$2B)</option>
                </select>
            </div>
            <div class="control-group">
                <label for="binSize">Bin Size (days)</label>
                <select id="binSize" onchange="updateCharts()">
                    <option value="5">5 days</option>
                    <option value="10" selected>10 days</option>
                    <option value="15">15 days</option>
                    <option value="20">20 days</option>
                </select>
            </div>
        </div>

        <div class="stats-grid" id="statsGrid">
            <!-- Stats populated by JavaScript -->
        </div>

        <div id="histogram"></div>
        <div id="boxplot"></div>

        <div class="footer">
            Generated by Downtrend Duration Analyzer | Peak/Trough Window: {peak_window} days | Min Depth: {min_depth}%
        </div>
    </div>

    <script>
        // Embedded data
        const allDowntrends = {downtrends_json};

        function filterData() {{
            const sector = document.getElementById('sectorFilter').value;
            const marketCap = document.getElementById('marketCapFilter').value;

            return allDowntrends.filter(d => {{
                const sectorMatch = sector === 'all' || d.sector === sector;
                const capMatch = marketCap === 'all' || d.market_cap_tier === marketCap;
                return sectorMatch && capMatch;
            }});
        }}

        function computeStats(data) {{
            if (data.length === 0) return null;

            const durations = data.map(d => d.duration_days).sort((a, b) => a - b);
            const n = durations.length;

            const sum = durations.reduce((a, b) => a + b, 0);
            const mean = sum / n;
            const median = n % 2 === 0
                ? (durations[n/2 - 1] + durations[n/2]) / 2
                : durations[Math.floor(n/2)];

            const p25 = durations[Math.floor(n * 0.25)];
            const p75 = durations[Math.floor(n * 0.75)];
            const p90 = durations[Math.floor(n * 0.90)];

            return {{ count: n, mean, median, p25, p75, p90 }};
        }}

        function updateStats(stats) {{
            const grid = document.getElementById('statsGrid');
            if (!stats) {{
                grid.innerHTML = '<div class="stat-card"><div class="stat-value">-</div><div class="stat-label">No Data</div></div>';
                return;
            }}

            grid.innerHTML = `
                <div class="stat-card">
                    <div class="stat-value">${{stats.count.toLocaleString()}}</div>
                    <div class="stat-label">Downtrends</div>
                </div>
                <div class="stat-card">
                    <div class="stat-value">${{Math.round(stats.median)}}</div>
                    <div class="stat-label">Median (days)</div>
                </div>
                <div class="stat-card">
                    <div class="stat-value">${{stats.mean.toFixed(1)}}</div>
                    <div class="stat-label">Mean (days)</div>
                </div>
                <div class="stat-card">
                    <div class="stat-value">${{stats.p25}}</div>
                    <div class="stat-label">P25 (days)</div>
                </div>
                <div class="stat-card">
                    <div class="stat-value">${{stats.p75}}</div>
                    <div class="stat-label">P75 (days)</div>
                </div>
                <div class="stat-card">
                    <div class="stat-value">${{stats.p90}}</div>
                    <div class="stat-label">P90 (days)</div>
                </div>
            `;
        }}

        function updateCharts() {{
            const data = filterData();
            const stats = computeStats(data);
            updateStats(stats);

            const binSize = parseInt(document.getElementById('binSize').value);
            const durations = data.map(d => d.duration_days);

            // Histogram
            const histTrace = {{
                x: durations,
                type: 'histogram',
                xbins: {{ size: binSize }},
                marker: {{
                    color: 'rgba(37, 99, 235, 0.7)',
                    line: {{ color: 'rgba(37, 99, 235, 1)', width: 1 }}
                }},
                hovertemplate: 'Duration: %{{x}} days<br>Count: %{{y}}<extra></extra>'
            }};

            const histLayout = {{
                title: 'Duration Distribution',
                xaxis: {{
                    title: 'Duration (trading days)',
                    dtick: binSize * 2
                }},
                yaxis: {{ title: 'Frequency' }},
                bargap: 0.05,
                shapes: stats ? [
                    {{ type: 'line', x0: stats.median, x1: stats.median, y0: 0, y1: 1, yref: 'paper', line: {{ color: 'red', width: 2, dash: 'dash' }} }},
                    {{ type: 'line', x0: stats.p90, x1: stats.p90, y0: 0, y1: 1, yref: 'paper', line: {{ color: 'orange', width: 2, dash: 'dot' }} }}
                ] : [],
                annotations: stats ? [
                    {{ x: stats.median, y: 1, yref: 'paper', text: 'Median', showarrow: false, yanchor: 'bottom' }},
                    {{ x: stats.p90, y: 1, yref: 'paper', text: 'P90', showarrow: false, yanchor: 'bottom' }}
                ] : []
            }};

            Plotly.newPlot('histogram', [histTrace], histLayout, {{ responsive: true }});

            // Box plot by market cap
            const marketCaps = ['Mega', 'Large', 'Mid', 'Small'];
            const boxTraces = marketCaps.map(cap => {{
                const capData = data.filter(d => d.market_cap_tier === cap);
                return {{
                    y: capData.map(d => d.duration_days),
                    name: cap,
                    type: 'box',
                    boxmean: true
                }};
            }});

            const boxLayout = {{
                title: 'Duration by Market Cap Tier',
                yaxis: {{ title: 'Duration (trading days)' }},
                showlegend: false
            }};

            Plotly.newPlot('boxplot', boxTraces, boxLayout, {{ responsive: true }});
        }}

        // Initial render
        updateCharts();
    </script>
</body>
</html>
"""


def load_analysis_data(input_path: Path) -> dict[str, Any]:
    """Load analysis JSON data."""
    with open(input_path) as f:
        return json.load(f)


def generate_sector_options(downtrends: list[dict]) -> str:
    """Generate HTML options for sector dropdown."""
    sectors = sorted(set(d.get("sector", "Unknown") for d in downtrends))
    return "\n".join(f'                    <option value="{s}">{s}</option>' for s in sectors)


def generate_html(data: dict[str, Any]) -> str:
    """Generate the HTML visualization."""
    params = data.get("parameters", {})
    summary = data.get("summary", {})
    downtrends = data.get("downtrends", [])

    sector_options = generate_sector_options(downtrends)

    html = HTML_TEMPLATE.format(
        analysis_date=data.get("analysis_date", "Unknown")[:10],
        lookback_years=params.get("lookback_years", "?"),
        total_downtrends=summary.get("total_downtrends", 0),
        peak_window=params.get("peak_window", 20),
        min_depth=params.get("min_depth_pct", 5.0),
        sector_options=sector_options,
        downtrends_json=json.dumps(downtrends),
    )

    return html


def find_latest_json(input_pattern: str, output_dir: Path) -> Path | None:
    """Find the latest matching JSON file."""
    if "*" in input_pattern:
        # Glob pattern
        base_dir = Path(input_pattern).parent
        pattern = Path(input_pattern).name

        if not base_dir.exists():
            base_dir = output_dir

        matches = sorted(base_dir.glob(pattern), reverse=True)
        return matches[0] if matches else None
    else:
        # Direct path
        path = Path(input_pattern)
        return path if path.exists() else None


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Generate interactive HTML histogram from downtrend analysis"
    )
    parser.add_argument(
        "--input",
        default="reports/downtrend_analysis_*.json",
        help="Input JSON file or glob pattern (default: reports/downtrend_analysis_*.json)",
    )
    parser.add_argument(
        "--output-dir",
        type=str,
        default="reports",
        help="Output directory for HTML file (default: reports)",
    )

    args = parser.parse_args()

    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    # Find input file
    input_path = find_latest_json(args.input, output_dir)
    if not input_path:
        print(f"Error: No input file found matching '{args.input}'", file=sys.stderr)
        sys.exit(1)

    print(f"Loading data from: {input_path}")

    # Load and process data
    data = load_analysis_data(input_path)
    html_content = generate_html(data)

    # Write output
    timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
    output_path = output_dir / f"downtrend_histogram_{timestamp}.html"

    output_path.write_text(html_content)
    print(f"HTML visualization saved to: {output_path}")


if __name__ == "__main__":
    main()

"""
Pytest configuration for downtrend-duration-analyzer tests.
"""

import sys
from pathlib import Path

# Add scripts directory to path for imports
scripts_dir = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(scripts_dir))

"""
Tests for analyze_downtrends.py

Tests peak/trough detection, downtrend identification, and statistics computation.
"""

import pandas as pd
from analyze_downtrends import (
    compute_statistics,
    detect_peaks_troughs,
    find_downtrends,
    get_market_cap_tier,
    group_statistics,
)


class TestGetMarketCapTier:
    """Tests for market cap tier classification."""

    def test_mega_cap(self):
        """Test mega cap classification."""
        assert get_market_cap_tier(250_000_000_000) == "Mega"
        assert get_market_cap_tier(200_000_000_000) == "Mega"

    def test_large_cap(self):
        """Test large cap classification."""
        assert get_market_cap_tier(100_000_000_000) == "Large"
        assert get_market_cap_tier(10_000_000_000) == "Large"

    def test_mid_cap(self):
        """Test mid cap classification."""
        assert get_market_cap_tier(5_000_000_000) == "Mid"
        assert get_market_cap_tier(2_000_000_000) == "Mid"

    def test_small_cap(self):
        """Test small cap classification."""
        assert get_market_cap_tier(1_000_000_000) == "Small"
        assert get_market_cap_tier(100_000_000) == "Small"

    def test_none_market_cap(self):
        """Test None market cap returns Unknown."""
        assert get_market_cap_tier(None) == "Unknown"


class TestDetectPeaksTroughs:
    """Tests for peak and trough detection algorithm."""

    def test_simple_peak_detection(self):
        """Test detection of a simple peak."""
        # Create data with a clear peak at index 5
        dates = pd.date_range("2024-01-01", periods=11)
        closes = [100, 102, 104, 106, 108, 110, 108, 106, 104, 102, 100]
        prices = pd.DataFrame({"date": dates, "close": closes})

        peaks, troughs = detect_peaks_troughs(prices, peak_window=2, trough_window=2)

        assert 5 in peaks  # Peak at 110

    def test_simple_trough_detection(self):
        """Test detection of a simple trough."""
        # Create data with a clear trough at index 5
        dates = pd.date_range("2024-01-01", periods=11)
        closes = [100, 98, 96, 94, 92, 90, 92, 94, 96, 98, 100]
        prices = pd.DataFrame({"date": dates, "close": closes})

        peaks, troughs = detect_peaks_troughs(prices, peak_window=2, trough_window=2)

        assert 5 in troughs  # Trough at 90

    def test_multiple_peaks_troughs(self):
        """Test detection of multiple peaks and troughs."""
        # Create oscillating data
        dates = pd.date_range("2024-01-01", periods=21)
        closes = [
            100,
            105,
            110,
            105,
            100,
            95,
            90,
            95,
            100,
            105,
            110,
            105,
            100,
            95,
            90,
            95,
            100,
            105,
            110,
            105,
            100,
        ]
        prices = pd.DataFrame({"date": dates, "close": closes})

        peaks, troughs = detect_peaks_troughs(prices, peak_window=2, trough_window=2)

        # Should find peaks at 110 and troughs at 90
        assert len(peaks) >= 2
        assert len(troughs) >= 2

    def test_insufficient_data(self):
        """Test that insufficient data returns empty lists."""
        dates = pd.date_range("2024-01-01", periods=5)
        closes = [100, 102, 104, 102, 100]
        prices = pd.DataFrame({"date": dates, "close": closes})

        peaks, troughs = detect_peaks_troughs(prices, peak_window=10, trough_window=10)

        assert peaks == []
        assert troughs == []


class TestFindDowntrends:
    """Tests for downtrend identification."""

    def test_identify_downtrend(self):
        """Test identification of a downtrend period."""
        dates = pd.date_range("2024-01-01", periods=21)
        # Peak at index 5 (110), trough at index 15 (90)
        closes = [
            100,
            102,
            104,
            106,
            108,
            110,
            108,
            106,
            104,
            102,
            100,
            98,
            96,
            94,
            92,
            90,
            92,
            94,
            96,
            98,
            100,
        ]
        prices = pd.DataFrame({"date": dates, "close": closes})

        peaks = [5]
        troughs = [15]

        downtrends = find_downtrends(prices, peaks, troughs, min_depth_pct=5.0, min_duration_days=3)

        assert len(downtrends) == 1
        assert downtrends[0]["duration_days"] == 10
        assert downtrends[0]["depth_pct"] < 0  # Negative because it's a decline

    def test_filter_shallow_downtrend(self):
        """Test that shallow downtrends are filtered out."""
        dates = pd.date_range("2024-01-01", periods=11)
        # Only 2% decline - should be filtered with 5% threshold
        closes = [100, 101, 102, 101, 100, 99, 98, 99, 100, 101, 102]
        prices = pd.DataFrame({"date": dates, "close": closes})

        peaks = [2]
        troughs = [6]

        downtrends = find_downtrends(prices, peaks, troughs, min_depth_pct=5.0, min_duration_days=1)

        assert len(downtrends) == 0

    def test_filter_short_downtrend(self):
        """Test that very short downtrends are filtered out."""
        dates = pd.date_range("2024-01-01", periods=11)
        closes = [100, 105, 110, 100, 90, 95, 100, 105, 110, 105, 100]
        prices = pd.DataFrame({"date": dates, "close": closes})

        peaks = [2]
        troughs = [4]

        # 2 days duration should be filtered with min_duration_days=3
        downtrends = find_downtrends(prices, peaks, troughs, min_depth_pct=5.0, min_duration_days=3)

        assert len(downtrends) == 0


class TestComputeStatistics:
    """Tests for statistics computation."""

    def test_empty_downtrends(self):
        """Test statistics for empty input."""
        stats = compute_statistics([])

        assert stats["total_downtrends"] == 0
        assert stats["median_duration_days"] == 0
        assert stats["mean_duration_days"] == 0

    def test_single_downtrend(self):
        """Test statistics for single downtrend."""
        downtrends = [{"duration_days": 20, "depth_pct": -10.0}]

        stats = compute_statistics(downtrends)

        assert stats["total_downtrends"] == 1
        assert stats["median_duration_days"] == 20
        assert stats["mean_duration_days"] == 20.0

    def test_multiple_downtrends(self):
        """Test statistics for multiple downtrends."""
        downtrends = [
            {"duration_days": 10},
            {"duration_days": 20},
            {"duration_days": 30},
            {"duration_days": 40},
            {"duration_days": 50},
        ]

        stats = compute_statistics(downtrends)

        assert stats["total_downtrends"] == 5
        assert stats["median_duration_days"] == 30
        assert stats["mean_duration_days"] == 30.0
        assert stats["p25_duration_days"] == 20
        assert stats["p75_duration_days"] == 40


class TestGroupStatistics:
    """Tests for grouped statistics computation."""

    def test_group_by_sector(self):
        """Test grouping by sector."""
        downtrends = [
            {"sector": "Technology", "duration_days": 10},
            {"sector": "Technology", "duration_days": 20},
            {"sector": "Healthcare", "duration_days": 30},
        ]

        grouped = group_statistics(downtrends, "sector")

        assert "Technology" in grouped
        assert "Healthcare" in grouped
        assert grouped["Technology"]["count"] == 2
        assert grouped["Healthcare"]["count"] == 1

    def test_group_by_market_cap(self):
        """Test grouping by market cap tier."""
        downtrends = [
            {"market_cap_tier": "Mega", "duration_days": 10},
            {"market_cap_tier": "Mega", "duration_days": 15},
            {"market_cap_tier": "Small", "duration_days": 30},
        ]

        grouped = group_statistics(downtrends, "market_cap_tier")

        assert "Mega" in grouped
        assert "Small" in grouped
        assert grouped["Mega"]["median_days"] == 12  # (10 + 15) / 2 rounded
        assert grouped["Small"]["median_days"] == 30

"""FMP /stable migration: stock list uses /stable/company-screener.

fetch_stock_list() used v3 /stock-screener (403 for keys issued after
2025-08-31). It now calls /stable/company-screener first with a v3 fallback;
both take the same params and return the same fields.
"""

from unittest.mock import MagicMock, patch

import analyze_downtrends


def _resp(status_code, payload):
    resp = MagicMock()
    resp.status_code = status_code
    resp.json.return_value = payload
    return resp


@patch("analyze_downtrends.requests.get")
def test_uses_stable_company_screener_first(mock_get):
    mock_get.return_value = _resp(
        200, [{"symbol": "XOM", "sector": "Energy", "marketCap": 500_000_000_000}]
    )
    stocks = analyze_downtrends.fetch_stock_list("key", sector="Energy")

    assert stocks[0]["symbol"] == "XOM"
    call = mock_get.call_args_list[0]
    assert call[0][0].endswith("/stable/company-screener")
    assert call[1]["params"]["sector"] == "Energy"
    assert call[1]["params"]["isActivelyTrading"] == "true"
    assert call[1]["params"]["limit"] == 500


@patch("analyze_downtrends.requests.get")
def test_falls_back_to_v3_stock_screener(mock_get):
    def fake_get(url, params=None, timeout=None):
        if url.endswith("/stable/company-screener"):
            return _resp(403, {})  # legacy/stable failure -> fallback
        return _resp(200, [{"symbol": "AAPL", "sector": "Technology"}])

    mock_get.side_effect = fake_get
    stocks = analyze_downtrends.fetch_stock_list("key")
    assert stocks[0]["symbol"] == "AAPL"
    urls = [c[0][0] for c in mock_get.call_args_list]
    assert any(u.endswith("/api/v3/stock-screener") for u in urls)


@patch("analyze_downtrends.requests.get")
def test_returns_empty_when_all_fail(mock_get):
    mock_get.return_value = _resp(403, {})
    assert analyze_downtrends.fetch_stock_list("key") == []

"""
Tests for generate_histogram_html.py

Tests HTML generation and data processing for visualization.
"""

import pytest
from generate_histogram_html import (
    find_latest_json,
    generate_html,
    generate_sector_options,
)


class TestGenerateSectorOptions:
    """Tests for sector dropdown generation."""

    def test_single_sector(self):
        """Test single sector option generation."""
        downtrends = [{"sector": "Technology"}]
        options = generate_sector_options(downtrends)

        assert "Technology" in options
        assert "<option" in options

    def test_multiple_sectors(self):
        """Test multiple sector options sorted alphabetically."""
        downtrends = [
            {"sector": "Technology"},
            {"sector": "Healthcare"},
            {"sector": "Energy"},
        ]
        options = generate_sector_options(downtrends)

        assert "Technology" in options
        assert "Healthcare" in options
        assert "Energy" in options
        # Check alphabetical order
        assert options.index("Energy") < options.index("Healthcare")
        assert options.index("Healthcare") < options.index("Technology")

    def test_duplicate_sectors(self):
        """Test that duplicate sectors are deduplicated."""
        downtrends = [
            {"sector": "Technology"},
            {"sector": "Technology"},
            {"sector": "Healthcare"},
        ]
        options = generate_sector_options(downtrends)

        # Count occurrences - should be 1 option tag for Technology
        assert options.count("<option") == 2  # Healthcare + Technology only

    def test_unknown_sector(self):
        """Test handling of missing sector data."""
        downtrends = [{"symbol": "ABC"}]  # No sector field
        options = generate_sector_options(downtrends)

        assert "Unknown" in options


class TestGenerateHtml:
    """Tests for HTML generation."""

    @pytest.fixture
    def sample_data(self):
        """Sample analysis data for testing."""
        return {
            "schema_version": "1.0",
            "analysis_date": "2026-03-28T07:00:00Z",
            "parameters": {
                "lookback_years": 5,
                "sector_filter": None,
                "peak_window": 20,
                "trough_window": 20,
                "min_depth_pct": 5.0,
            },
            "summary": {
                "total_downtrends": 100,
                "median_duration_days": 18,
                "mean_duration_days": 24.5,
                "p25_duration_days": 10,
                "p75_duration_days": 32,
                "p90_duration_days": 55,
            },
            "downtrends": [
                {
                    "symbol": "AAPL",
                    "sector": "Technology",
                    "market_cap_tier": "Mega",
                    "peak_date": "2025-01-15",
                    "trough_date": "2025-02-10",
                    "duration_days": 18,
                    "depth_pct": -12.5,
                },
                {
                    "symbol": "JNJ",
                    "sector": "Healthcare",
                    "market_cap_tier": "Large",
                    "peak_date": "2025-02-01",
                    "trough_date": "2025-02-20",
                    "duration_days": 14,
                    "depth_pct": -8.3,
                },
            ],
        }

    def test_html_contains_plotly(self, sample_data):
        """Test that generated HTML includes Plotly.js."""
        html = generate_html(sample_data)

        assert "plotly" in html.lower()
        assert "cdn.plot.ly" in html

    def test_html_contains_analysis_date(self, sample_data):
        """Test that analysis date is included."""
        html = generate_html(sample_data)

        assert "2026-03-28" in html

    def test_html_contains_total_count(self, sample_data):
        """Test that total downtrend count is included."""
        html = generate_html(sample_data)

        assert "100" in html  # total_downtrends

    def test_html_contains_sector_options(self, sample_data):
        """Test that sector filter options are generated."""
        html = generate_html(sample_data)

        assert "Technology" in html
        assert "Healthcare" in html

    def test_html_contains_downtrend_data(self, sample_data):
        """Test that downtrend data is embedded as JSON."""
        html = generate_html(sample_data)

        assert "AAPL" in html
        assert "duration_days" in html

    def test_html_is_valid_structure(self, sample_data):
        """Test that HTML has valid structure."""
        html = generate_html(sample_data)

        assert html.startswith("<!DOCTYPE html>")
        assert "<html" in html
        assert "</html>" in html
        assert "<head>" in html
        assert "</head>" in html
        assert "<body>" in html
        assert "</body>" in html


class TestFindLatestJson:
    """Tests for JSON file finding logic."""

    def test_direct_path(self, tmp_path):
        """Test finding a directly specified file."""
        json_file = tmp_path / "test.json"
        json_file.write_text('{"test": true}')

        result = find_latest_json(str(json_file), tmp_path)

        assert result == json_file

    def test_glob_pattern(self, tmp_path):
        """Test finding files with glob pattern."""
        # Create multiple JSON files
        (tmp_path / "analysis_2026-01-01.json").write_text('{"date": "2026-01-01"}')
        (tmp_path / "analysis_2026-02-01.json").write_text('{"date": "2026-02-01"}')

        result = find_latest_json(str(tmp_path / "analysis_*.json"), tmp_path)

        # Should find the latest (alphabetically last) file
        assert result is not None
        assert "2026-02-01" in str(result)

    def test_no_match(self, tmp_path):
        """Test behavior when no files match."""
        result = find_latest_json(str(tmp_path / "nonexistent_*.json"), tmp_path)

        assert result is None

    def test_nonexistent_direct_path(self, tmp_path):
        """Test behavior when direct path doesn't exist."""
        result = find_latest_json(str(tmp_path / "nonexistent.json"), tmp_path)

        assert result is None