Sector Analyst

#!/usr/bin/env python3
"""Sector rotation analysis from TraderMonty's public CSV data.

.. note:: Uses ``from __future__ import annotations`` for Python 3.9 compat.

Fetches sector_summary.csv (and uptrend_ratio_timeseries.csv for freshness
check) from GitHub, then produces sector rankings, risk-regime scoring,
overbought/oversold flags, and market-cycle phase estimation.

Data Source:
  https://github.com/tradermonty/uptrend-dashboard

Dependencies: stdlib only (urllib, csv, json, argparse, dataclasses).
"""

from __future__ import annotations

import argparse
import csv
import io
import json
import math
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass
from datetime import date, datetime

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

SECTOR_CSV_URL = (
    "https://raw.githubusercontent.com/tradermonty/uptrend-dashboard/main/data/sector_summary.csv"
)
UPTREND_CSV_URL = (
    "https://raw.githubusercontent.com/tradermonty/uptrend-dashboard"
    "/main/data/uptrend_ratio_timeseries.csv"
)

# Sector classification — aligned with uptrend-analyzer's
# sector_rotation_calculator.py (L31-47).
CYCLICAL_SECTORS = [
    "Technology",
    "Consumer Cyclical",
    "Communication Services",
    "Financial",
    "Industrials",
]
DEFENSIVE_SECTORS = [
    "Utilities",
    "Consumer Defensive",
    "Healthcare",
    "Real Estate",
]
COMMODITY_SECTORS = [
    "Energy",
    "Basic Materials",
]

# Overbought / oversold thresholds (uptrend ratio scale).
OVERBOUGHT_THRESHOLD = 0.37
OVERSOLD_THRESHOLD = 0.097

# Freshness: warn if data is older than this many days.
FRESHNESS_MAX_DAYS = 5

# Cycle phase definitions: which sectors lead/lag in each phase.
CYCLE_PHASES = {
    "early": {
        "leaders": ["Technology", "Consumer Cyclical", "Industrials", "Financial"],
        "laggards": ["Utilities", "Consumer Defensive", "Healthcare"],
    },
    "mid": {
        "leaders": ["Technology", "Industrials", "Consumer Cyclical", "Energy"],
        "laggards": ["Utilities", "Consumer Defensive"],
    },
    "late": {
        "leaders": ["Energy", "Basic Materials", "Healthcare"],
        "laggards": ["Technology", "Consumer Cyclical", "Industrials"],
    },
    "recession": {
        "leaders": ["Utilities", "Consumer Defensive", "Healthcare"],
        "laggards": ["Technology", "Consumer Cyclical", "Industrials", "Financial"],
    },
}


# ---------------------------------------------------------------------------
# Dataclass
# ---------------------------------------------------------------------------


@dataclass
class SectorData:
    """Parsed sector row."""

    sector: str
    ratio: float
    ma_10: float | None
    trend: str
    slope: float | None
    status: str


# ---------------------------------------------------------------------------
# CSV Fetching
# ---------------------------------------------------------------------------


# Required columns for sector_summary.csv validation.
REQUIRED_COLUMNS = {"Sector", "Ratio"}
EXPECTED_COLUMNS = {"Sector", "Ratio", "10MA", "Trend", "Slope", "Status"}


def fetch_csv(url: str, timeout: int = 30) -> list[dict]:
    """Fetch a CSV from *url* and return rows as list of dicts."""
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "sector-analyst/1.0"})
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            text = resp.read().decode("utf-8")
    except urllib.error.URLError as exc:
        print(f"ERROR: Failed to fetch {url}: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

    reader = csv.DictReader(io.StringIO(text))
    return list(reader)


def validate_columns(raw_rows: list[dict]) -> None:
    """Validate that required columns exist in CSV data.

    Raises ValueError if required columns are missing.
    Prints a warning if optional expected columns are missing.
    """
    if not raw_rows:
        return
    actual = set(raw_rows[0].keys())
    missing_required = REQUIRED_COLUMNS - actual
    if missing_required:
        raise ValueError(f"Missing required columns: {sorted(missing_required)}")
    missing_optional = EXPECTED_COLUMNS - REQUIRED_COLUMNS - actual
    if missing_optional:
        print(
            f"WARNING: Missing optional columns: {sorted(missing_optional)}",
            file=sys.stderr,
        )


def parse_sector_rows(raw_rows: list[dict]) -> list[SectorData]:
    """Convert raw CSV dicts to SectorData, skipping invalid rows."""
    result: list[SectorData] = []
    for row in raw_rows:
        sector = row.get("Sector", "").strip()
        if not sector:
            continue
        ratio_str = row.get("Ratio", "").strip()
        if not ratio_str:
            continue
        try:
            ratio = float(ratio_str)
        except ValueError:
            continue

        ma_10 = _safe_float(row.get("10MA", ""))
        slope = _safe_float(row.get("Slope", ""))
        trend = row.get("Trend", "").strip()
        status = row.get("Status", "").strip()

        result.append(SectorData(sector, ratio, ma_10, trend, slope, status))
    return result


def check_freshness(uptrend_csv_url: str, timeout: int = 30) -> dict | None:
    """Check data freshness using max(date) from uptrend timeseries CSV.

    Returns dict with 'date', 'is_fresh', 'warning' or None on failure.
    Fetch failures are treated as non-fatal (returns None with no ERROR log).
    """
    try:
        req = urllib.request.Request(uptrend_csv_url, headers={"User-Agent": "sector-analyst/1.0"})
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            text = resp.read().decode("utf-8")
        reader = csv.DictReader(io.StringIO(text))
        rows = list(reader)
    except (urllib.error.URLError, OSError):
        return None

    if not rows:
        return None

    dates: list[str] = []
    for row in rows:
        d = row.get("date", "").strip()
        if d:
            dates.append(d)

    if not dates:
        return None

    max_date_str = max(dates)
    try:
        max_date = datetime.strptime(max_date_str, "%Y-%m-%d").date()
    except ValueError:
        return None

    days_old = (date.today() - max_date).days
    is_fresh = days_old <= FRESHNESS_MAX_DAYS
    warning = None if is_fresh else f"Data is {days_old} days old (latest: {max_date_str})"

    return {"date": max_date_str, "is_fresh": is_fresh, "warning": warning}


# ---------------------------------------------------------------------------
# Analysis Functions
# ---------------------------------------------------------------------------


def rank_sectors(sectors: list[SectorData]) -> list[dict]:
    """Rank sectors by ratio descending."""
    sorted_sectors = sorted(sectors, key=lambda s: s.ratio, reverse=True)
    return [
        {
            "rank": i + 1,
            "sector": s.sector,
            "ratio": s.ratio,
            "ratio_pct": round(s.ratio * 100, 1),
            "trend": s.trend,
            "slope": s.slope,
            "status": s.status,
        }
        for i, s in enumerate(sorted_sectors)
    ]


def analyze_groups(sectors: list[SectorData]) -> dict:
    """Analyze cyclical/defensive/commodity groups and determine risk regime."""
    sector_map = {s.sector: s for s in sectors}

    cyclical_ratios = _get_group_ratios(sector_map, CYCLICAL_SECTORS)
    defensive_ratios = _get_group_ratios(sector_map, DEFENSIVE_SECTORS)
    commodity_ratios = _get_group_ratios(sector_map, COMMODITY_SECTORS)

    cyclical_avg = _avg(cyclical_ratios) if cyclical_ratios else None
    defensive_avg = _avg(defensive_ratios) if defensive_ratios else None
    commodity_avg = _avg(commodity_ratios) if commodity_ratios else None

    if cyclical_avg is None or defensive_avg is None:
        return {
            "cyclical_avg": cyclical_avg,
            "defensive_avg": defensive_avg,
            "commodity_avg": commodity_avg,
            "difference": None,
            "score": 50,
            "regime": "incomplete data",
            "late_cycle_flag": False,
            "divergence_flag": False,
        }

    difference = cyclical_avg - defensive_avg
    base_score = _difference_to_score(difference)

    # Commodity adjustment
    late_cycle_flag = False
    commodity_penalty = 0
    if commodity_avg is not None:
        if commodity_avg > cyclical_avg and commodity_avg > defensive_avg:
            late_cycle_flag = True
            excess = commodity_avg - max(cyclical_avg, defensive_avg)
            commodity_penalty = -10 if excess > 0.10 else -5

    # Divergence detection
    divergence_result = _calculate_group_divergence(sector_map)
    divergence_penalty = divergence_result.get("divergence_penalty", 0)

    score = round(min(100, max(0, base_score + commodity_penalty + divergence_penalty)))
    regime = _score_to_regime(score)

    return {
        "cyclical_avg": round(cyclical_avg, 4),
        "cyclical_avg_pct": round(cyclical_avg * 100, 1),
        "defensive_avg": round(defensive_avg, 4),
        "defensive_avg_pct": round(defensive_avg * 100, 1),
        "commodity_avg": round(commodity_avg, 4) if commodity_avg is not None else None,
        "commodity_avg_pct": round(commodity_avg * 100, 1) if commodity_avg is not None else None,
        "difference": round(difference, 4),
        "difference_pct": round(difference * 100, 1),
        "score": score,
        "regime": regime,
        "late_cycle_flag": late_cycle_flag,
        "commodity_penalty": commodity_penalty,
        "divergence_flag": divergence_result.get("divergence_flag", False),
        "divergence_penalty": divergence_penalty,
    }


def identify_overbought_oversold(
    sectors: list[SectorData],
) -> tuple[list[dict], list[dict]]:
    """Identify overbought (ratio > 0.37) and oversold (ratio < 0.097) sectors."""
    overbought: list[dict] = []
    oversold: list[dict] = []
    for s in sectors:
        if s.ratio > OVERBOUGHT_THRESHOLD:
            overbought.append(
                {"sector": s.sector, "ratio": s.ratio, "ratio_pct": round(s.ratio * 100, 1)}
            )
        elif s.ratio < OVERSOLD_THRESHOLD:
            oversold.append(
                {"sector": s.sector, "ratio": s.ratio, "ratio_pct": round(s.ratio * 100, 1)}
            )
    return overbought, oversold


def analyze_trends(sectors: list[SectorData]) -> dict:
    """Analyze trend distribution across sectors."""
    up = [s.sector for s in sectors if s.trend.lower() == "up"]
    down = [s.sector for s in sectors if s.trend.lower() == "down"]
    return {
        "uptrend_count": len(up),
        "downtrend_count": len(down),
        "uptrend_sectors": up,
        "downtrend_sectors": down,
    }


def estimate_cycle_phase(sectors: list[SectorData]) -> dict:
    """Estimate market cycle phase based on sector performance patterns.

    Scoring per phase:
      - Leader match in top ranks: weight 0.4
      - Laggard match in bottom ranks: weight 0.3
      - Trend direction alignment: weight 0.3
    """
    if not sectors:
        return {"phase": "unknown", "confidence": "low", "scores": {}, "evidence": []}

    sorted_by_ratio = sorted(sectors, key=lambda s: s.ratio, reverse=True)
    n = len(sorted_by_ratio)
    top_half = {s.sector for s in sorted_by_ratio[: n // 2 + 1]}
    bottom_half = {s.sector for s in sorted_by_ratio[n // 2 :]}
    trend_map = {s.sector: s.trend.lower() for s in sectors}

    scores: dict[str, float] = {}
    evidence: list[str] = []

    for phase_name, phase_def in CYCLE_PHASES.items():
        leaders = phase_def["leaders"]
        laggards = phase_def["laggards"]

        # Leader match: how many expected leaders are in top half
        leader_hits = sum(1 for s in leaders if s in top_half)
        leader_score = leader_hits / len(leaders) if leaders else 0

        # Laggard match: how many expected laggards are in bottom half
        laggard_hits = sum(1 for s in laggards if s in bottom_half)
        laggard_score = laggard_hits / len(laggards) if laggards else 0

        # Trend alignment: leaders trending up, laggards trending down
        trend_hits = 0
        trend_total = 0
        for s in leaders:
            if s in trend_map:
                trend_total += 1
                if trend_map[s] == "up":
                    trend_hits += 1
        for s in laggards:
            if s in trend_map:
                trend_total += 1
                if trend_map[s] == "down":
                    trend_hits += 1
        trend_score = trend_hits / trend_total if trend_total else 0

        total = leader_score * 0.4 + laggard_score * 0.3 + trend_score * 0.3
        scores[phase_name] = round(total * 100, 1)

    # Determine best phase
    sorted_phases = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    best_phase = sorted_phases[0][0]
    best_score = sorted_phases[0][1]
    second_score = sorted_phases[1][1] if len(sorted_phases) > 1 else 0

    gap = best_score - second_score
    if gap > 20:
        confidence = "high"
    elif gap > 10:
        confidence = "moderate"
    else:
        confidence = "low"

    # Build evidence
    phase_def = CYCLE_PHASES[best_phase]
    matched_leaders = [s for s in phase_def["leaders"] if s in top_half]
    matched_laggards = [s for s in phase_def["laggards"] if s in bottom_half]
    if matched_leaders:
        evidence.append(f"Leaders in top ranks: {', '.join(matched_leaders)}")
    if matched_laggards:
        evidence.append(f"Laggards in bottom ranks: {', '.join(matched_laggards)}")

    return {
        "phase": best_phase,
        "confidence": confidence,
        "scores": scores,
        "evidence": evidence,
    }


# ---------------------------------------------------------------------------
# Output Formatters
# ---------------------------------------------------------------------------


def format_human(
    ranking: list[dict],
    groups: dict,
    overbought: list[dict],
    oversold: list[dict],
    trends: dict,
    cycle: dict,
    freshness: dict | None,
) -> str:
    """Format analysis results as human-readable markdown."""
    lines: list[str] = []
    lines.append(f"# Sector Rotation Analysis — {date.today().isoformat()}")
    lines.append("")

    # Freshness warning
    if freshness:
        if freshness.get("warning"):
            lines.append(f"> **WARNING**: {freshness['warning']}")
            lines.append("")
        else:
            lines.append(f"> Data as of: {freshness['date']}")
            lines.append("")

    # Risk Regime
    lines.append("## Risk Regime")
    lines.append("")
    regime = groups.get("regime", "N/A").upper()
    score = groups.get("score", "N/A")
    lines.append(f"**{regime}** (score: {score}/100)")
    lines.append("")
    if groups.get("cyclical_avg") is not None:
        lines.append(f"- Cyclical avg: {groups['cyclical_avg_pct']}%")
        lines.append(f"- Defensive avg: {groups['defensive_avg_pct']}%")
        lines.append(f"- Difference: {groups['difference_pct']}pp")
        if groups.get("commodity_avg") is not None:
            lines.append(f"- Commodity avg: {groups['commodity_avg_pct']}%")
        if groups.get("late_cycle_flag"):
            lines.append(
                "- **Late Cycle Flag**: Commodity sectors leading both cyclical and defensive"
            )
        if groups.get("divergence_flag"):
            lines.append("- **Divergence Flag**: High intra-group spread detected")
    lines.append("")

    # Cycle Phase
    lines.append("## Cycle Phase Estimate")
    lines.append("")
    phase = cycle.get("phase", "unknown").replace("_", " ").title()
    conf = cycle.get("confidence", "N/A")
    lines.append(f"**{phase}** (confidence: {conf})")
    lines.append("")
    if cycle.get("scores"):
        for p, s in sorted(cycle["scores"].items(), key=lambda x: x[1], reverse=True):
            marker = " ←" if p == cycle["phase"] else ""
            lines.append(f"- {p.title()}: {s}{marker}")
    lines.append("")
    if cycle.get("evidence"):
        lines.append("Evidence:")
        for e in cycle["evidence"]:
            lines.append(f"- {e}")
        lines.append("")

    # Sector Ranking
    lines.append("## Sector Ranking (by uptrend ratio)")
    lines.append("")
    lines.append("| Rank | Sector | Ratio | Trend | Status |")
    lines.append("|------|--------|-------|-------|--------|")
    for r in ranking:
        lines.append(
            f"| {r['rank']} | {r['sector']} | {r['ratio_pct']}% | {r['trend']} | {r['status']} |"
        )
    lines.append("")

    # Trends
    lines.append("## Trend Summary")
    lines.append("")
    lines.append(f"- Uptrending: {trends['uptrend_count']} sectors")
    lines.append(f"- Downtrending: {trends['downtrend_count']} sectors")
    lines.append("")

    # Overbought / Oversold
    if overbought or oversold:
        lines.append("## Overbought / Oversold")
        lines.append("")
        if overbought:
            lines.append("**Overbought** (ratio > 37%):")
            for s in overbought:
                lines.append(f"- {s['sector']}: {s['ratio_pct']}%")
        if oversold:
            lines.append("**Oversold** (ratio < 9.7%):")
            for s in oversold:
                lines.append(f"- {s['sector']}: {s['ratio_pct']}%")
        lines.append("")

    lines.append("---")
    lines.append(f"*Generated: {datetime.now().isoformat(timespec='seconds')}*")
    return "\n".join(lines)


def format_json(
    ranking: list[dict],
    groups: dict,
    overbought: list[dict],
    oversold: list[dict],
    trends: dict,
    cycle: dict,
    freshness: dict | None,
) -> str:
    """Format analysis results as JSON."""
    data = {
        "meta": {
            "generated_at": datetime.now().isoformat(timespec="seconds"),
            "date": date.today().isoformat(),
            "freshness": freshness,
        },
        "groups": groups,
        "cycle_phase": cycle,
        "ranking": ranking,
        "overbought": overbought,
        "oversold": oversold,
        "trends": trends,
    }
    return json.dumps(data, indent=2, ensure_ascii=False)


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------


def _safe_float(value: str) -> float | None:
    """Parse float, return None on failure."""
    try:
        return float(value.strip())
    except (ValueError, AttributeError):
        return None


def _avg(values: list[float]) -> float:
    """Simple average."""
    return sum(values) / len(values)


def _get_group_ratios(sector_map: dict[str, SectorData], sector_names: list[str]) -> list[float]:
    """Extract ratios for a group of sectors."""
    ratios: list[float] = []
    for name in sector_names:
        sector = sector_map.get(name)
        if sector is not None:
            ratios.append(sector.ratio)
    return ratios


def _difference_to_score(diff: float) -> float:
    """Map cyclical-defensive difference to score (0-100).

    Aligned with sector_rotation_calculator.py L171-189.
    """
    if diff > 0.15:
        return min(100, 90 + (diff - 0.15) / 0.10 * 10)
    elif diff > 0.05:
        return 70 + (diff - 0.05) / 0.10 * 19
    elif diff > -0.05:
        return 45 + (diff + 0.05) / 0.10 * 24
    elif diff > -0.15:
        return 20 + (diff + 0.15) / 0.10 * 24
    else:
        return max(0, 19 + (diff + 0.15) / 0.10 * 19)


def _score_to_regime(score: int) -> str:
    """Map score to risk regime label."""
    if score >= 90:
        return "strong risk-on"
    elif score >= 70:
        return "risk-on"
    elif score >= 45:
        return "balanced"
    elif score >= 20:
        return "defensive tilt"
    else:
        return "strong risk-off"


def _calculate_group_divergence(sector_map: dict[str, SectorData]) -> dict:
    """Detect intra-group divergence.

    Aligned with sector_rotation_calculator.py L209-232.
    """
    cyclical_div = _analyze_group(sector_map, CYCLICAL_SECTORS)
    defensive_div = _analyze_group(sector_map, DEFENSIVE_SECTORS)

    flag = cyclical_div["flagged"] or defensive_div["flagged"]
    penalty = -5 if flag else 0

    return {
        "divergence_flag": flag,
        "divergence_penalty": penalty,
        "cyclical_divergence": cyclical_div,
        "defensive_divergence": defensive_div,
    }


def _analyze_group(sector_map: dict[str, SectorData], sector_names: list[str]) -> dict:
    """Analyze divergence within a sector group."""
    ratios: list[float] = []
    trends: list[str] = []
    names_with_data: list[str] = []

    for name in sector_names:
        sector = sector_map.get(name)
        if sector is not None:
            ratios.append(sector.ratio)
            trends.append(sector.trend.lower())
            names_with_data.append(name)

    if len(ratios) < 2:
        return {
            "flagged": False,
            "std_dev": None,
            "spread": None,
            "outliers": [],
            "trend_dissenters": [],
        }

    mean = sum(ratios) / len(ratios)
    variance = sum((r - mean) ** 2 for r in ratios) / len(ratios)
    std_dev = math.sqrt(variance)
    spread = max(ratios) - min(ratios)

    # Outlier detection
    outliers: list[dict] = []
    for name, ratio in zip(names_with_data, ratios):
        if abs(ratio - mean) > 1.5 * std_dev and std_dev > 0:
            outliers.append({"sector": name, "ratio": ratio, "deviation": round(ratio - mean, 4)})

    # Trend dissenter detection
    trend_dissenters: list[dict] = []
    if trends:
        up_count = sum(1 for t in trends if t == "up")
        down_count = sum(1 for t in trends if t == "down")
        majority = "up" if up_count >= down_count else "down"

        for name, trend in zip(names_with_data, trends):
            if trend and trend != majority:
                trend_dissenters.append({"sector": name, "trend": trend, "majority": majority})

    flagged = std_dev > 0.08 or spread > 0.20 or len(trend_dissenters) > 0

    return {
        "flagged": flagged,
        "std_dev": round(std_dev, 4),
        "spread": round(spread, 4),
        "outliers": outliers,
        "trend_dissenters": trend_dissenters,
    }


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main() -> None:
    """Entry point."""
    parser = argparse.ArgumentParser(
        description="Sector rotation analysis from TraderMonty CSV data"
    )
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument(
        "--output-dir", default="reports/", help="Output directory (default: reports/)"
    )
    parser.add_argument("--save", action="store_true", help="Save output to file")
    parser.add_argument("--url", default=SECTOR_CSV_URL, help="Custom sector CSV URL")
    parser.add_argument(
        "--uptrend-url", default=UPTREND_CSV_URL, help="Custom uptrend timeseries CSV URL"
    )
    parser.add_argument("--timeout", type=int, default=30, help="HTTP timeout in seconds")
    args = parser.parse_args()

    # 1. Freshness check
    print("Checking data freshness...", file=sys.stderr)
    freshness = check_freshness(args.uptrend_url, timeout=args.timeout)
    if freshness is None:
        print("WARNING: Could not check data freshness (continuing)", file=sys.stderr)
    elif freshness.get("warning"):
        print(f"WARNING: {freshness['warning']}", file=sys.stderr)
    else:
        print(f"Data is fresh (latest: {freshness['date']})", file=sys.stderr)

    # 2. Fetch and parse sector data
    print("Fetching sector summary...", file=sys.stderr)
    raw_rows = fetch_csv(args.url, timeout=args.timeout)
    try:
        validate_columns(raw_rows)
    except ValueError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc
    sectors = parse_sector_rows(raw_rows)

    if not sectors:
        print("ERROR: No valid sector data found", file=sys.stderr)
        raise SystemExit(1)

    print(f"Parsed {len(sectors)} sectors", file=sys.stderr)

    # 3. Analyze
    ranking = rank_sectors(sectors)
    groups = analyze_groups(sectors)
    overbought, oversold = identify_overbought_oversold(sectors)
    trends = analyze_trends(sectors)
    cycle = estimate_cycle_phase(sectors)

    # 4. Format output
    if args.json:
        output = format_json(ranking, groups, overbought, oversold, trends, cycle, freshness)
    else:
        output = format_human(ranking, groups, overbought, oversold, trends, cycle, freshness)

    print(output)

    # 5. Save to file
    if args.save:
        import os

        os.makedirs(args.output_dir, exist_ok=True)
        ext = ".json" if args.json else ".md"
        filename = f"sector_rotation_{date.today().isoformat()}{ext}"
        filepath = os.path.join(args.output_dir, filename)
        with open(filepath, "w") as f:
            f.write(output)
        print(f"Saved to {filepath}", file=sys.stderr)


if __name__ == "__main__":
    main()

"""Path setup for sector-analyst tests."""

import sys
from pathlib import Path

_tests_dir = Path(__file__).resolve().parent
_scripts_dir = _tests_dir.parent

# Add scripts/ to sys.path so tests can import analyze_sector_rotation
sys.path.insert(0, str(_scripts_dir))
# Add tests/ to sys.path so tests can import helpers
sys.path.insert(0, str(_tests_dir))

"""Test data factories for sector-analyst tests."""


def make_sector_row(
    sector: str,
    ratio: float = 0.20,
    ma_10: float = 0.19,
    trend: str = "Up",
    slope: float = 0.005,
    status: str = "Above MA",
) -> dict:
    """Create a single sector CSV row dict."""
    return {
        "Sector": sector,
        "Ratio": str(ratio),
        "10MA": str(ma_10),
        "Trend": trend,
        "Slope": str(slope),
        "Status": status,
    }


ALL_SECTORS = [
    "Technology",
    "Consumer Cyclical",
    "Communication Services",
    "Financial",
    "Industrials",
    "Utilities",
    "Consumer Defensive",
    "Healthcare",
    "Real Estate",
    "Energy",
    "Basic Materials",
]


def make_full_sector_set(
    ratio: float = 0.20,
    trend: str = "Up",
    slope: float = 0.005,
) -> list[dict]:
    """Create a full set of 11 sector rows with uniform defaults."""
    return [make_sector_row(s, ratio=ratio, trend=trend, slope=slope) for s in ALL_SECTORS]


def make_early_cycle_scenario() -> list[dict]:
    """Early cycle: cyclical sectors high, defensive sectors low."""
    leaders = {
        "Technology",
        "Communication Services",
        "Industrials",
        "Consumer Cyclical",
        "Financial",
    }
    rows = []
    for s in ALL_SECTORS:
        if s in leaders:
            rows.append(make_sector_row(s, ratio=0.35, trend="Up", slope=0.010))
        else:
            rows.append(make_sector_row(s, ratio=0.10, trend="Down", slope=-0.005))
    return rows


def make_late_cycle_scenario() -> list[dict]:
    """Late cycle: energy/materials high, tech/consumer cyclical low."""
    commodity_leaders = {"Energy", "Basic Materials"}
    cyclical_low = {"Technology", "Consumer Cyclical"}
    rows = []
    for s in ALL_SECTORS:
        if s in commodity_leaders:
            rows.append(make_sector_row(s, ratio=0.40, trend="Up", slope=0.012))
        elif s in cyclical_low:
            rows.append(make_sector_row(s, ratio=0.10, trend="Down", slope=-0.008))
        else:
            rows.append(make_sector_row(s, ratio=0.20, trend="Up", slope=0.002))
    return rows


def make_recession_scenario() -> list[dict]:
    """Recession: defensive sectors high, cyclical/commodity low."""
    defensives = {"Utilities", "Healthcare", "Consumer Defensive"}
    rows = []
    for s in ALL_SECTORS:
        if s in defensives:
            rows.append(make_sector_row(s, ratio=0.40, trend="Up", slope=0.010))
        else:
            rows.append(make_sector_row(s, ratio=0.08, trend="Down", slope=-0.008))
    return rows

"""Unit tests for analyze_sector_rotation.py."""

import json
from datetime import date, timedelta
from unittest.mock import MagicMock, patch

import pytest
from analyze_sector_rotation import (
    CYCLICAL_SECTORS,
    DEFENSIVE_SECTORS,
    SectorData,
    analyze_groups,
    analyze_trends,
    check_freshness,
    estimate_cycle_phase,
    fetch_csv,
    format_human,
    format_json,
    identify_overbought_oversold,
    parse_sector_rows,
    rank_sectors,
    validate_columns,
)
from helpers import (
    ALL_SECTORS,
    make_early_cycle_scenario,
    make_full_sector_set,
    make_late_cycle_scenario,
    make_recession_scenario,
    make_sector_row,
)


# ---------------------------------------------------------------------------
# TestParseSectorRows
# ---------------------------------------------------------------------------
class TestParseSectorRows:
    """Tests for CSV row parsing into SectorData."""

    def test_basic_parse(self):
        rows = [make_sector_row("Technology", ratio=0.25, trend="Up", slope=0.008)]
        result = parse_sector_rows(rows)
        assert len(result) == 1
        assert result[0].sector == "Technology"
        assert result[0].ratio == pytest.approx(0.25)
        assert result[0].trend == "Up"
        assert result[0].slope == pytest.approx(0.008)

    def test_ma_10_parsed_from_10MA_column(self):
        """Verify that ma_10 is read from the '10MA' column (actual CSV name)."""
        row = {
            "Sector": "Technology",
            "Ratio": "0.25",
            "10MA": "0.23",
            "Trend": "Up",
            "Slope": "0.005",
            "Status": "Normal",
        }
        result = parse_sector_rows([row])
        assert result[0].ma_10 == pytest.approx(0.23)

    def test_full_set_parses_all_11(self):
        rows = make_full_sector_set()
        result = parse_sector_rows(rows)
        assert len(result) == 11

    def test_missing_ratio_skips_row(self):
        row = make_sector_row("Technology")
        row["Ratio"] = ""
        result = parse_sector_rows([row])
        assert len(result) == 0

    def test_non_numeric_ratio_skips_row(self):
        row = make_sector_row("Technology")
        row["Ratio"] = "N/A"
        result = parse_sector_rows([row])
        assert len(result) == 0

    def test_missing_sector_skips_row(self):
        row = make_sector_row("Technology")
        row["Sector"] = ""
        result = parse_sector_rows([row])
        assert len(result) == 0

    def test_missing_optional_fields_use_defaults(self):
        row = {"Sector": "Technology", "Ratio": "0.30"}
        result = parse_sector_rows([row])
        assert len(result) == 1
        assert result[0].ma_10 is None
        assert result[0].slope is None
        assert result[0].status == ""

    def test_empty_list_returns_empty(self):
        assert parse_sector_rows([]) == []


# ---------------------------------------------------------------------------
# TestValidateColumns
# ---------------------------------------------------------------------------
class TestValidateColumns:
    """Tests for CSV column validation."""

    def test_valid_columns_pass(self):
        rows = make_full_sector_set()
        validate_columns(rows)  # Should not raise

    def test_missing_required_column_raises(self):
        rows = [{"Ratio": "0.25", "10MA": "0.23"}]  # Missing "Sector"
        with pytest.raises(ValueError, match="Sector"):
            validate_columns(rows)

    def test_missing_optional_column_warns(self, capsys):
        rows = [{"Sector": "Technology", "Ratio": "0.25"}]  # Missing optional cols
        validate_columns(rows)  # Should not raise
        captured = capsys.readouterr()
        assert "WARNING" in captured.err

    def test_empty_rows_no_error(self):
        validate_columns([])  # Should not raise


# ---------------------------------------------------------------------------
# TestRankSectors
# ---------------------------------------------------------------------------
class TestRankSectors:
    """Tests for ratio-descending ranking."""

    def test_rank_order(self):
        sectors = [
            SectorData("A", 0.10, None, "Up", None, ""),
            SectorData("B", 0.30, None, "Up", None, ""),
            SectorData("C", 0.20, None, "Up", None, ""),
        ]
        ranked = rank_sectors(sectors)
        assert ranked[0]["sector"] == "B"
        assert ranked[1]["sector"] == "C"
        assert ranked[2]["sector"] == "A"
        assert ranked[0]["rank"] == 1
        assert ranked[2]["rank"] == 3

    def test_rank_includes_ratio_pct(self):
        sectors = [SectorData("A", 0.25, None, "Up", None, "")]
        ranked = rank_sectors(sectors)
        assert ranked[0]["ratio_pct"] == pytest.approx(25.0)

    def test_empty_input(self):
        assert rank_sectors([]) == []


# ---------------------------------------------------------------------------
# TestAnalyzeGroups
# ---------------------------------------------------------------------------
class TestAnalyzeGroups:
    """Tests for cyclical/defensive group analysis and risk regime."""

    def test_balanced_scenario(self):
        sectors = parse_sector_rows(make_full_sector_set(ratio=0.20))
        result = analyze_groups(sectors)
        assert result["cyclical_avg"] == pytest.approx(0.20)
        assert result["defensive_avg"] == pytest.approx(0.20)
        assert result["difference"] == pytest.approx(0.0)
        assert result["regime"] == "balanced"

    def test_risk_on_scenario(self):
        rows = []
        for s in ALL_SECTORS:
            if s in CYCLICAL_SECTORS:
                rows.append(make_sector_row(s, ratio=0.35))
            else:
                rows.append(make_sector_row(s, ratio=0.10))
        sectors = parse_sector_rows(rows)
        result = analyze_groups(sectors)
        assert result["cyclical_avg"] > result["defensive_avg"]
        assert result["regime"] in ("risk-on", "strong risk-on")

    def test_risk_off_scenario(self):
        rows = []
        for s in ALL_SECTORS:
            if s in DEFENSIVE_SECTORS:
                rows.append(make_sector_row(s, ratio=0.35))
            else:
                rows.append(make_sector_row(s, ratio=0.05))
        sectors = parse_sector_rows(rows)
        result = analyze_groups(sectors)
        assert result["defensive_avg"] > result["cyclical_avg"]
        assert result["regime"] in ("defensive tilt", "strong risk-off")

    def test_commodity_avg_calculated(self):
        sectors = parse_sector_rows(make_full_sector_set(ratio=0.20))
        result = analyze_groups(sectors)
        assert result["commodity_avg"] == pytest.approx(0.20)

    def test_late_cycle_flag(self):
        rows = make_late_cycle_scenario()
        sectors = parse_sector_rows(rows)
        result = analyze_groups(sectors)
        assert result["late_cycle_flag"] is True

    def test_score_range_0_to_100(self):
        for ratio in [0.01, 0.10, 0.20, 0.30, 0.45]:
            rows = make_full_sector_set(ratio=ratio)
            sectors = parse_sector_rows(rows)
            result = analyze_groups(sectors)
            assert 0 <= result["score"] <= 100

    def test_divergence_flag_high_spread(self):
        """High spread within cyclical group triggers divergence."""
        rows = make_full_sector_set(ratio=0.20)
        # Make Technology very high and Industrials very low
        for r in rows:
            if r["Sector"] == "Technology":
                r["Ratio"] = "0.45"
            elif r["Sector"] == "Industrials":
                r["Ratio"] = "0.05"
        sectors = parse_sector_rows(rows)
        result = analyze_groups(sectors)
        assert result["divergence_flag"] is True


# ---------------------------------------------------------------------------
# TestOverboughtOversold
# ---------------------------------------------------------------------------
class TestOverboughtOversold:
    """Tests for overbought/oversold identification."""

    def test_overbought_threshold(self):
        sectors = [SectorData("A", 0.40, None, "Up", None, "")]
        ob, os_ = identify_overbought_oversold(sectors)
        assert len(ob) == 1
        assert ob[0]["sector"] == "A"

    def test_oversold_threshold(self):
        sectors = [SectorData("A", 0.05, None, "Down", None, "")]
        ob, os_ = identify_overbought_oversold(sectors)
        assert len(os_) == 1
        assert os_[0]["sector"] == "A"

    def test_normal_range_no_flags(self):
        sectors = [SectorData("A", 0.20, None, "Up", None, "")]
        ob, os_ = identify_overbought_oversold(sectors)
        assert len(ob) == 0
        assert len(os_) == 0

    def test_boundary_values(self):
        # Exactly at threshold
        sectors = [
            SectorData("A", 0.37, None, "Up", None, ""),
            SectorData("B", 0.097, None, "Down", None, ""),
        ]
        ob, os_ = identify_overbought_oversold(sectors)
        # At threshold = not triggered (must exceed)
        assert len(ob) == 0
        assert len(os_) == 0


# ---------------------------------------------------------------------------
# TestAnalyzeTrends
# ---------------------------------------------------------------------------
class TestAnalyzeTrends:
    """Tests for trend analysis."""

    def test_all_up(self):
        sectors = parse_sector_rows(make_full_sector_set(trend="Up"))
        result = analyze_trends(sectors)
        assert result["uptrend_count"] == 11
        assert result["downtrend_count"] == 0

    def test_mixed_trends(self):
        rows = make_early_cycle_scenario()
        sectors = parse_sector_rows(rows)
        result = analyze_trends(sectors)
        assert result["uptrend_count"] > 0
        assert result["downtrend_count"] > 0
        assert result["uptrend_count"] + result["downtrend_count"] == 11


# ---------------------------------------------------------------------------
# TestEstimateCyclePhase
# ---------------------------------------------------------------------------
class TestEstimateCyclePhase:
    """Tests for market cycle phase estimation."""

    def test_early_cycle(self):
        sectors = parse_sector_rows(make_early_cycle_scenario())
        result = estimate_cycle_phase(sectors)
        assert result["phase"] == "early"

    def test_late_cycle(self):
        sectors = parse_sector_rows(make_late_cycle_scenario())
        result = estimate_cycle_phase(sectors)
        assert result["phase"] == "late"

    def test_recession(self):
        sectors = parse_sector_rows(make_recession_scenario())
        result = estimate_cycle_phase(sectors)
        assert result["phase"] == "recession"

    def test_confidence_high_when_clear(self):
        sectors = parse_sector_rows(make_early_cycle_scenario())
        result = estimate_cycle_phase(sectors)
        assert result["confidence"] in ("high", "moderate")

    def test_scores_dict_has_all_phases(self):
        sectors = parse_sector_rows(make_full_sector_set())
        result = estimate_cycle_phase(sectors)
        for phase in ("early", "mid", "late", "recession"):
            assert phase in result["scores"]

    def test_evidence_list_not_empty(self):
        sectors = parse_sector_rows(make_early_cycle_scenario())
        result = estimate_cycle_phase(sectors)
        assert len(result["evidence"]) > 0


# ---------------------------------------------------------------------------
# TestFormatHuman
# ---------------------------------------------------------------------------
class TestFormatHuman:
    """Tests for human-readable output formatting."""

    def test_contains_required_sections(self):
        sectors = parse_sector_rows(make_full_sector_set())
        ranking = rank_sectors(sectors)
        groups = analyze_groups(sectors)
        ob, os_ = identify_overbought_oversold(sectors)
        trends = analyze_trends(sectors)
        cycle = estimate_cycle_phase(sectors)
        output = format_human(ranking, groups, ob, os_, trends, cycle, None)
        assert "Sector Ranking" in output
        assert "Risk Regime" in output
        assert "Cycle Phase" in output

    def test_freshness_warning_included(self):
        sectors = parse_sector_rows(make_full_sector_set())
        ranking = rank_sectors(sectors)
        groups = analyze_groups(sectors)
        ob, os_ = identify_overbought_oversold(sectors)
        trends = analyze_trends(sectors)
        cycle = estimate_cycle_phase(sectors)
        freshness = {"date": "2025-01-01", "is_fresh": False, "warning": "Data is 60 days old"}
        output = format_human(ranking, groups, ob, os_, trends, cycle, freshness)
        assert "WARNING" in output or "Data is 60 days old" in output


# ---------------------------------------------------------------------------
# TestFormatJson
# ---------------------------------------------------------------------------
class TestFormatJson:
    """Tests for JSON output formatting."""

    def test_valid_json(self):
        sectors = parse_sector_rows(make_full_sector_set())
        ranking = rank_sectors(sectors)
        groups = analyze_groups(sectors)
        ob, os_ = identify_overbought_oversold(sectors)
        trends = analyze_trends(sectors)
        cycle = estimate_cycle_phase(sectors)
        output = format_json(ranking, groups, ob, os_, trends, cycle, None)
        data = json.loads(output)
        assert "ranking" in data
        assert "groups" in data
        assert "cycle_phase" in data

    def test_json_has_meta_section(self):
        sectors = parse_sector_rows(make_full_sector_set())
        ranking = rank_sectors(sectors)
        groups = analyze_groups(sectors)
        ob, os_ = identify_overbought_oversold(sectors)
        trends = analyze_trends(sectors)
        cycle = estimate_cycle_phase(sectors)
        output = format_json(ranking, groups, ob, os_, trends, cycle, None)
        data = json.loads(output)
        assert "meta" in data
        assert "generated_at" in data["meta"]


# ---------------------------------------------------------------------------
# TestFetchCsv
# ---------------------------------------------------------------------------
class TestFetchCsv:
    """Tests for CSV fetching with mocked urllib."""

    @patch("analyze_sector_rotation.urllib.request.urlopen")
    def test_fetch_csv_success(self, mock_urlopen):
        csv_content = (
            "Sector,Ratio,MA_10,Trend,Slope,Status\nTechnology,0.25,0.24,Up,0.005,Above MA\n"
        )
        mock_response = MagicMock()
        mock_response.read.return_value = csv_content.encode("utf-8")
        mock_response.__enter__ = lambda s: s
        mock_response.__exit__ = MagicMock(return_value=False)
        mock_urlopen.return_value = mock_response

        rows = fetch_csv("https://example.com/data.csv")
        assert len(rows) == 1
        assert rows[0]["Sector"] == "Technology"

    @patch("analyze_sector_rotation.urllib.request.urlopen")
    def test_fetch_csv_error(self, mock_urlopen):
        from urllib.error import URLError

        mock_urlopen.side_effect = URLError("connection failed")
        with pytest.raises(SystemExit):
            fetch_csv("https://example.com/data.csv")

    @patch("analyze_sector_rotation.urllib.request.urlopen")
    def test_fetch_csv_empty_response(self, mock_urlopen):
        csv_content = "Sector,Ratio,MA_10,Trend,Slope,Status\n"
        mock_response = MagicMock()
        mock_response.read.return_value = csv_content.encode("utf-8")
        mock_response.__enter__ = lambda s: s
        mock_response.__exit__ = MagicMock(return_value=False)
        mock_urlopen.return_value = mock_response

        rows = fetch_csv("https://example.com/data.csv")
        assert len(rows) == 0


# ---------------------------------------------------------------------------
# TestCheckFreshness
# ---------------------------------------------------------------------------
class TestCheckFreshness:
    """Tests for data freshness checking via direct urllib (no fetch_csv)."""

    @staticmethod
    def _make_mock_response(csv_text: str) -> MagicMock:
        mock_resp = MagicMock()
        mock_resp.read.return_value = csv_text.encode("utf-8")
        mock_resp.__enter__ = lambda s: s
        mock_resp.__exit__ = MagicMock(return_value=False)
        return mock_resp

    @patch("analyze_sector_rotation.urllib.request.urlopen")
    def test_fresh_data(self, mock_urlopen):
        today = date.today()
        yesterday = today - timedelta(days=1)
        csv_text = (
            "date,value\n"
            f"{today - timedelta(days=10)},0.1\n"
            f"{yesterday},0.2\n"
            f"{today - timedelta(days=5)},0.15\n"
        )
        mock_urlopen.return_value = self._make_mock_response(csv_text)
        result = check_freshness("https://example.com/ts.csv")
        assert result["date"] == str(yesterday)
        assert result["is_fresh"] is True
        assert result["warning"] is None

    @patch("analyze_sector_rotation.urllib.request.urlopen")
    def test_stale_data(self, mock_urlopen):
        old_date = date.today() - timedelta(days=10)
        csv_text = f"date,value\n{old_date - timedelta(days=5)},0.1\n{old_date},0.2\n"
        mock_urlopen.return_value = self._make_mock_response(csv_text)
        result = check_freshness("https://example.com/ts.csv")
        assert result["is_fresh"] is False
        assert result["warning"] is not None

    @patch("analyze_sector_rotation.urllib.request.urlopen")
    def test_fetch_failure_returns_none(self, mock_urlopen):
        from urllib.error import URLError

        mock_urlopen.side_effect = URLError("connection failed")
        result = check_freshness("https://example.com/ts.csv")
        assert result is None

    @patch("analyze_sector_rotation.urllib.request.urlopen")
    def test_no_error_log_on_fetch_failure(self, mock_urlopen, capsys):
        """Freshness fetch failure should not produce ERROR log."""
        from urllib.error import URLError

        mock_urlopen.side_effect = URLError("connection failed")
        check_freshness("https://example.com/ts.csv")
        captured = capsys.readouterr()
        assert "ERROR" not in captured.err

    @patch("analyze_sector_rotation.urllib.request.urlopen")
    def test_order_independent(self, mock_urlopen):
        """Max date is found regardless of row order."""
        today = date.today()
        csv_text = (
            "date,value\n"
            f"{today - timedelta(days=3)},0.1\n"
            f"{today},0.2\n"
            f"{today - timedelta(days=7)},0.15\n"
        )
        mock_urlopen.return_value = self._make_mock_response(csv_text)
        result = check_freshness("https://example.com/ts.csv")
        assert result["date"] == str(today)

    @patch("analyze_sector_rotation.urllib.request.urlopen")
    def test_empty_timeseries(self, mock_urlopen):
        csv_text = "date,value\n"
        mock_urlopen.return_value = self._make_mock_response(csv_text)
        result = check_freshness("https://example.com/ts.csv")
        assert result is None