"""Export and chart revenue/EPS estimate revisions for one target period.

https://vizualstocks.com/api/tutorials/estimate-revisions-python
Python 3.10+. Install requests, pandas, matplotlib. Set VIZUALSTOCKS_API_KEY.
"""
import argparse
import math
import os
from pathlib import Path
import re
from datetime import datetime, timezone
import json

import pandas as pd
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

BASE_URL = "https://api.vizualstocks.com/v1"


def make_session():
    key = os.environ.get("VIZUALSTOCKS_API_KEY", "").strip()
    if not key: raise ValueError("Set VIZUALSTOCKS_API_KEY in this terminal first.")
    session = requests.Session()
    session.headers.update({"X-API-Key": key, "Accept": "application/json"})
    retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 502, 503, 504], allowed_methods=["GET"], respect_retry_after_header=True)
    session.mount("https://", HTTPAdapter(max_retries=retry))
    return session


def api_get(session, path, **params):
    response = session.get(BASE_URL + path, params=params, timeout=(10, 60), allow_redirects=False)
    if 300 <= response.status_code < 400: raise ValueError("Unexpected API redirect.")
    response.raise_for_status()
    return response.json()


def revision_frame(payload, period, frequency):
    groups = [item for item in payload.get("data", []) if item.get("frequency") == frequency]
    matches = [item for group in groups for item in group.get("periods", []) if item.get("normalizedPeriod") == period]
    if len(matches) != 1 or not matches[0].get("history"):
        raise ValueError("No unique history for that target period. Run --list-periods first.")
    rows = []
    for point in matches[0]["history"]:
        day = point.get("collectedDate")
        if not isinstance(day, str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}", day): raise ValueError("Invalid collection date.")
        datetime.strptime(day, "%Y-%m-%d")
        row = {"collectedDate": day}
        for field in ["revenueEstimate", "epsEstimate"]:
            value = point.get(field)
            if value is not None and (isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value)):
                raise ValueError(f"Invalid value for {field}.")
            row[field] = value
        rows.append(row)
    frame = pd.DataFrame(rows).sort_values("collectedDate").reset_index(drop=True)
    if frame.collectedDate.duplicated().any(): raise ValueError("Duplicate collection dates for the same target period.")
    for field in ["revenueEstimate", "epsEstimate"]: frame[field] = frame[field].astype("Float64")
    if frame[["revenueEstimate", "epsEstimate"]].isna().all().all(): raise ValueError("No estimate values returned.")
    frame.insert(0, "normalizedPeriod", period)
    frame.insert(0, "frequency", frequency)
    frame.insert(0, "reportingCurrency", payload.get("reportingCurrency"))
    frame.insert(0, "ticker", payload["ticker"])
    frame["revenueChange"] = frame.revenueEstimate.diff().round(8)
    frame["epsChange"] = frame.epsEstimate.diff().round(8)
    return frame


def save_chart(frame, output):
    import matplotlib
    matplotlib.use("Agg")
    import matplotlib.pyplot as plt
    import matplotlib.dates as mdates
    dates = pd.to_datetime(frame.collectedDate)
    currency = frame.reportingCurrency.iloc[0] or "currency not returned"
    fig, axes = plt.subplots(2, 1, figsize=(10, 6), sharex=True, layout="constrained")
    for axis, field, scale, label, color in [
        (axes[0], "revenueEstimate", 1_000_000, f"Revenue estimate ({currency} millions)", "#3576b9"),
        (axes[1], "epsEstimate", 1, f"Normalized EPS ({currency}/share)", "#17846d")]:
        axis.plot(dates, frame[field].to_numpy(dtype=float, na_value=float("nan")) / scale, marker="o", markersize=2, color=color, linewidth=1.4)
        axis.set_ylabel(label)
        axis.grid(axis="y", alpha=0.2)
        axis.spines[["top", "right"]].set_visible(False)
        if frame[field].isna().all(): axis.text(0.5, 0.5, "Not returned", ha="center", transform=axis.transAxes)
    locator = mdates.AutoDateLocator(minticks=3, maxticks=8)
    axes[1].xaxis.set_major_locator(locator)
    axes[1].xaxis.set_major_formatter(mdates.ConciseDateFormatter(locator))
    axes[1].set_xlabel("Collection date · one fixed target period")
    fig.suptitle(f"{frame.ticker.iloc[0]} estimates for {frame.normalizedPeriod.iloc[0]} ({frame.frequency.iloc[0]})")
    try:
        with Path(output).open("xb") as file: fig.savefig(file, format="png", dpi=160)
    finally: plt.close(fig)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--ticker", default="PLTR")
    parser.add_argument("--frequency", choices=["quarterly", "annual"], default="quarterly")
    parser.add_argument("--period", help="Exact normalized period, e.g. Q3'26 or 2027")
    parser.add_argument("--list-periods", action="store_true")
    parser.add_argument("--output-dir", default="pltr-estimate-revisions")
    args = parser.parse_args()
    try:
        ticker = args.ticker.strip().upper()
        if not re.fullmatch(r"[A-Z0-9][A-Z0-9.-]{0,19}", ticker): raise ValueError("Use a valid stock ticker.")
        if not args.list_periods and not re.fullmatch(r"Q[1-4]'\d{2}" if args.frequency == "quarterly" else r"\d{4}", args.period or ""):
            raise ValueError("Choose --period from --list-periods using the same frequency.")
        destination = Path(args.output_dir)
        if not args.list_periods and destination.exists(): raise ValueError("Choose a new --output-dir; existing exports are never overwritten.")
        with make_session() as session:
            if args.list_periods:
                payload = api_get(session, f"/companies/{ticker}/estimate-revisions", frequency=args.frequency, order="desc", limit=1)
                for group in payload.get("data", []):
                    for period in group.get("periods", []): print(period["normalizedPeriod"])
                return
            payload = api_get(session, f"/companies/{ticker}/estimate-revisions", frequency=args.frequency, normalizedPeriod=args.period, order="asc")
        frame = revision_frame(payload, args.period, args.frequency)
        destination.mkdir(parents=True, exist_ok=False)
        frame.to_csv(destination / "estimate-revisions.csv", index=False, na_rep="", mode="x")
        save_chart(frame, destination / "estimate-revisions.png")
        with (destination / "readme.json").open("x", encoding="utf-8") as file:
            json.dump({"ticker": ticker, "normalizedPeriod": args.period, "frequency": args.frequency,
                       "retrievedAtUtc": datetime.now(timezone.utc).isoformat(), "observations": len(frame),
                       "notes": "Collection date is not an original analyst publication timestamp. Changes compare adjacent returned observations of the same target period. EPS changes are absolute, not percentage returns."}, file, indent=2)
        print(f"Saved {len(frame)} revisions for {args.period} in {destination}.")
    except requests.HTTPError as error: parser.exit(1, f"API HTTP {error.response.status_code}. Check key, access and quota.\n")
    except requests.RequestException: parser.exit(1, "API connection failed after retries. Try again later.\n")
    except (ValueError, OSError, KeyError, TypeError) as error: parser.exit(1, f"{error}\n")


if __name__ == "__main__": main()