"""Build dated stock-ranking inputs from three VizualStocks history endpoints. https://vizualstocks.com/api/tutorials/historical-analyst-data-python Python 3.10+. Install requests. Set VIZUALSTOCKS_API_KEY for live requests. --sample uses explicitly synthetic companies, without a key or network requests. """ import argparse import csv from datetime import date, datetime, timedelta, timezone from html import escape import json import math import os from pathlib import Path import re import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry BASE_URL = "https://api.vizualstocks.com/v1" ROUTES = {"estimates": "estimate-revisions", "consensus": "analyst-rating-consensus-history", "guidance": "company-guidance-history"} RATING_FIELDS = ("strongBuy", "buy", "hold", "sell", "strongSell") TARGET_FIELDS = ("currency", "consensusPriceTarget", "medianPriceTarget", "lowPriceTarget", "highPriceTarget", "priceTargetContributors") SIGNALS = ("revenueRevisionPct", "targetRevisionPct", "buyShareChangePp") def day(value): if not isinstance(value, str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value): raise ValueError("Expected a YYYY-MM-DD date.") return date.fromisoformat(value) def number(value): if value is None: return None if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): raise ValueError("Expected a finite JSON number or null.") return value def identity(value): return (value or "").strip().casefold() def make_session(): key = os.environ.get("VIZUALSTOCKS_API_KEY", "").strip() if not key: raise ValueError("Set VIZUALSTOCKS_API_KEY or use --sample.") 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 fetch_histories(session, ticker, frequency, period, decision_dates=(), lookback_days=90): """Fetch full histories plus authoritative guidance states at each unique boundary.""" payloads, errors = {}, {} tasks = [] for name, route in ROUTES.items(): params = {"order": "asc"} if name == "estimates": params.update(frequency=frequency, normalizedPeriod=period) if name == "guidance": params.update(metric="Revenue", periodType=frequency) tasks.append((name, route, params)) boundaries = {day(value) - timedelta(days=offset) for value in decision_dates for offset in (1, 1 + lookback_days)} for boundary in sorted(boundaries): stamp = boundary.isoformat() tasks.append(("guidance@" + stamp, ROUTES["guidance"], dict(order="asc", metric="Revenue", periodType=frequency, asOfDate=stamp))) for name, route, params in tasks: try: response = session.get(f"{BASE_URL}/companies/{ticker}/{route}", params=params, timeout=(10, 60), allow_redirects=False) if 300 <= response.status_code < 400: raise ValueError("Unexpected redirect; credentials were not forwarded.") response.raise_for_status() payload = response.json() if not isinstance(payload, dict) or identity(payload.get("ticker")) != ticker.casefold(): raise ValueError("Response ticker does not match the request.") if not isinstance(payload.get("data"), list): raise ValueError("Response data is not a list.") payloads[name] = payload except requests.HTTPError as error: if error.response.status_code == 401: raise ValueError("API HTTP 401: check your API key.") from None errors[name] = f"HTTP {error.response.status_code}" except requests.RequestException: errors[name] = "Connection failed after retries" except ValueError as error: errors[name] = str(error) return payloads, errors def estimate_series(payload, frequency, period): matches = [row for group in payload.get("data", []) if group["frequency"] == frequency for row in group["periods"] if row["normalizedPeriod"] == period] if len(matches) > 1: raise ValueError("Ambiguous estimate target period.") return matches[0] if matches else {"history": [], "periodEndDate": None} def latest(rows, date_field, cutoff): eligible = [row for row in rows if day(row[date_field]) <= cutoff] if not eligible: return None last_date = max(row[date_field] for row in eligible) found = [row for row in eligible if row[date_field] == last_date] if len(found) != 1: raise ValueError(f"Multiple observations on {last_date}; resolve before ranking.") return found[0] def pair(rows, date_field, cutoff, baseline, max_age_days): current, previous = latest(rows, date_field, cutoff), latest(rows, date_field, baseline) status = "ready" if current is None or previous is None: status = "missing_history" elif max_age_days is not None and any( (boundary - day(row[date_field])).days > max_age_days for row, boundary in [(current, cutoff), (previous, baseline)] ): status = "observation_gap" return current, previous, status def change(current, previous): current, previous = number(current), number(previous) return current - previous if current is not None and previous is not None else None def percent_change(current, previous): delta = change(current, previous) return delta / previous * 100 if delta is not None and previous > 0 else None def buy_share(row): counts = [number(row.get(field)) for field in RATING_FIELDS] if any(value is None for value in counts): return None if any(value < 0 or int(value) != value for value in counts): raise ValueError("Rating counts must be nonnegative integers.") return 100 * sum(counts[:2]) / sum(counts) if sum(counts) else None ACTIVE_GUIDANCE_STATUSES = frozenset({ "issued", "updated", "updated_raised", "updated_lowered", "updated_narrowed", "updated_widened", "reaffirmed", }) def guidance_at(rows, cutoff, period_end, frequency, currency, basis=None): if not period_end or not currency: return None, "unmatched_period_or_currency" eligible = [row for row in rows if day(row["announcementDate"]) <= cutoff and identity(row.get("metric")) == "revenue" and identity(row.get("scope")) == "consolidated" and row.get("periodEndDate") == period_end and identity(row.get("periodType")) == frequency and identity(row.get("unit")) == identity(currency) and (basis is None or identity(row.get("basis")) == identity(basis))] groups = {} for row in eligible: groups.setdefault(identity(row.get("basis")), []).append(row) if len(groups) != 1: return None, "no_matching_guidance" if not groups else "choose_guidance_basis" row = latest(next(iter(groups.values())), "announcementDate", cutoff) state = identity(row.get("status")) return row, state if state in ACTIVE_GUIDANCE_STATUSES or state == "withdrawn" else "unknown_status" def guidance_bounds(row, state): if row is None or state not in ACTIVE_GUIDANCE_STATUSES: return None, None point, low, high = (number(row.get(key)) for key in ("value", "low", "high")) if point is not None: return point, point if low is not None and high is not None and low <= high: return low, high return None, None def build_row(ticker, payloads, decision_date, frequency, period, lookback_days=90, max_age_days=None, guidance_basis=None): """Daily inputs: use only the calendar day BEFORE the decision date or earlier.""" decision = day(decision_date) cutoff, baseline = decision - timedelta(days=1), decision - timedelta(days=1 + lookback_days) estimates = payloads.get("estimates", {}) target = estimate_series(estimates, frequency, period) period_end = target.get("periodEndDate") if period_end: day(period_end) currency = estimates.get("reportingCurrency") row = dict(ticker=ticker, decisionDate=decision_date, cutoffDate=cutoff.isoformat(), baselineCutoffDate=baseline.isoformat(), frequency=frequency, normalizedPeriod=period, periodEndDate=period_end, reportingCurrency=currency, lookbackDays=lookback_days) consensus = payloads.get("consensus", {}).get("data", []) # The unified endpoint joins ratings and targets by snapshot date. Keep each # signal's observation dates independent when one side of a snapshot is null. target_history = [point for point in consensus if any(point.get(field) is not None for field in TARGET_FIELDS)] rating_history = [point for point in consensus if any(point.get(field) is not None for field in RATING_FIELDS)] selected = {} for name, history, date_field in [ ("estimate", target.get("history", []), "collectedDate"), ("target", target_history, "asOfDate"), ("rating", rating_history, "asOfDate"), ]: current, previous, status = pair(history, date_field, cutoff, baseline, max_age_days) selected[name] = (current or {}, previous or {}) row[name + "Status"] = status for label, point, boundary in [("Date", current, cutoff), ("BaselineDate", previous, baseline)]: row[name + label] = point[date_field] if point else None row[name + label + "AgeDays"] = (boundary - day(point[date_field])).days if point else None current, previous = selected["estimate"] row.update(revenueEstimate=number(current.get("revenueEstimate")), epsEstimate=number(current.get("epsEstimate")), revenueRevisionPct=None, epsChange=None) if period_end and day(period_end) <= cutoff: row["estimateStatus"] = "target_period_ended" if row["estimateStatus"] == "ready": row["revenueRevisionPct"] = percent_change(current.get("revenueEstimate"), previous.get("revenueEstimate")) row["epsChange"] = change(current.get("epsEstimate"), previous.get("epsEstimate")) current, previous = selected["target"] row.update(targetMean=number(current.get("consensusPriceTarget")), targetMedian=number(current.get("medianPriceTarget")), targetCurrency=current.get("currency"), targetAnalystCount=number(current.get("priceTargetContributors")), targetLow=number(current.get("lowPriceTarget")), targetHigh=number(current.get("highPriceTarget")), targetRevisionPct=None, targetSpreadPct=None) if row["targetStatus"] == "ready": if current.get("currency") and identity(current.get("currency")) == identity(previous.get("currency")): row["targetRevisionPct"] = percent_change(current.get("consensusPriceTarget"), previous.get("consensusPriceTarget")) else: row["targetStatus"] = "currency_mismatch_or_missing" low, high, mean = row["targetLow"], row["targetHigh"], row["targetMean"] if low is not None and high is not None and mean is not None and high >= low and mean > 0: row["targetSpreadPct"] = 100 * (high - low) / mean current, previous = selected["rating"] row.update(buySharePct=buy_share(current), buyShareChangePp=None) if row["ratingStatus"] == "ready": row["buyShareChangePp"] = change(buy_share(current), buy_share(previous)) # Dated guidance responses are authoritative. Full history can omit a conflicting # announcement; blindly carrying forward its preceding row would revive stale guidance. current_guidance = payloads.get("guidance@" + cutoff.isoformat()) older_guidance = payloads.get("guidance@" + baseline.isoformat()) guidance, state = guidance_at((current_guidance or {}).get("data", []), cutoff, period_end, frequency, currency, guidance_basis) if current_guidance is None: state = "snapshot_unavailable" guide = guidance or {} low, high = guidance_bounds(guidance, state) older, older_state = guidance_at((older_guidance or {}).get("data", []), baseline, period_end, frequency, currency, guide.get("basis") if guidance else guidance_basis) old_low, old_high = guidance_bounds(older, older_state) midpoint = (low + high) / 2 if low is not None and high is not None else None old_midpoint = (old_low + old_high) / 2 if old_low is not None and old_high is not None else None row.update(guidanceState=state, guidanceDate=guide.get("announcementDate"), guidanceBasis=guide.get("basis"), guidanceLow=low, guidanceHigh=high, guidanceMidpointDerived=midpoint, guidanceBaselineDate=(older or {}).get("announcementDate"), guidanceMidpointChangePct=percent_change(midpoint, old_midpoint), guidanceValueText=guide.get("valueText"), guidanceSourceUrl=guide.get("sourceUrl"), consensusVsGuidance="unavailable") if midpoint is not None and row["revenueEstimate"] is not None: # Explicit opt-in: caller verified the chosen basis against the consensus definition. if not guidance_basis or identity(guidance_basis) in ("not specified", "unknown"): row["consensusVsGuidance"] = "review_basis" elif row["estimateStatus"] in ("observation_gap", "target_period_ended"): row["consensusVsGuidance"] = row["estimateStatus"] else: value = row["revenueEstimate"] row["consensusVsGuidance"] = "below" if value < low else "above" if value > high else "inside" return row def rank_rows(rows, signals=SIGNALS): """Equal-weight 0–100 percentile positions, calculated separately for each decision date.""" if not signals or len(set(signals)) != len(signals) or any(signal not in SIGNALS for signal in signals): raise ValueError("Choose distinct signals from: " + ", ".join(SIGNALS)) for decision in sorted({row["decisionDate"] for row in rows}): group = [row for row in rows if row["decisionDate"] == decision] eligible = [row for row in group if all(row.get(signal) is not None for signal in signals)] for row in group: row.update(score=None, rank=None, rankedCompanyCount=len(eligible), rankingStatus="missing_signals" if row not in eligible else "need_two_companies") if len(eligible) < 2: continue for row in eligible: positions = [] for signal in signals: values = [item[signal] for item in eligible] less = sum(value < row[signal] for value in values) tied = sum(value == row[signal] for value in values) positions.append(100 * (less + (tied - 1) / 2) / (len(values) - 1)) row["score"] = sum(positions) / len(positions) row["rankingStatus"] = "ranked" for row in eligible: row["rank"] = 1 + sum(item["score"] > row["score"] for item in eligible) return sorted(rows, key=lambda row: (row["decisionDate"], row["rank"] or math.inf, row["ticker"])) def coverage_rows(ticker, payloads, errors, frequency, period, cutoff): result = [] for name in ROUTES: payload = payloads.get(name, {}) rows = estimate_series(payload, frequency, period).get("history", []) if name == "estimates" else payload.get("data", []) field = "collectedDate" if name == "estimates" else "announcementDate" if name == "guidance" else "asOfDate" dates = sorted(row[field] for row in rows if day(row[field]) <= cutoff) result.append(dict(ticker=ticker, dataset=name, throughDate=cutoff.isoformat(), firstDate=dates[0] if dates else None, lastDate=dates[-1] if dates else None, observationCount=len(dates), status=errors.get(name, "available" if dates else "no_history"))) return result def sample_payloads(): """Synthetic, deliberately small fixtures, never presented as reported company data.""" result = {} for ticker, revenue, target, buys, guidance in [ ("EXAMPLEA", 1100, 110, 7, (1080, 1120)), ("EXAMPLEB", 1040, 102, 6, (1050, 1070)), ("EXAMPLEC", 980, 95, 4, (930, 970)), ]: result[ticker] = { "estimates": {"ticker": ticker, "reportingCurrency": "USD", "data": [{"frequency": "annual", "periods": [{ "normalizedPeriod": "2027", "periodEndDate": "2027-12-31", "history": [ {"collectedDate": "2026-07-01", "revenueEstimate": 1000, "epsEstimate": 2}, {"collectedDate": "2026-09-29", "revenueEstimate": revenue, "epsEstimate": 2.2}]}]}]}, "consensus": {"ticker": ticker, "data": [ {"asOfDate": when, "strongBuy": 0, "buy": count, "hold": 10 - count, "sell": 0, "strongSell": 0, "consensusRating": "buy" if count >= 5 else "hold", "consensusRatingValue": 3 + count / 10, "ratingsContributors": 10, "currency": "USD", "consensusPriceTarget": price, "medianPriceTarget": price, "lowPriceTarget": price - 20, "highPriceTarget": price + 20, "priceTargetContributors": 10} for when, count, price in [("2026-07-01", 5, 100), ("2026-09-29", buys, target)]]}, "guidance": {"ticker": ticker, "data": [ {"announcementDate": when, "metric": "Revenue", "scope": "Consolidated", "periodType": "annual", "periodEndDate": "2027-12-31", "basis": "GAAP", "unit": "USD", "value": None, "low": bounds[0], "high": bounds[1], "valueText": None, "status": status, "sourceUrl": None} for when, bounds, status in [("2026-07-01", (980, 1020), "issued"), ("2026-09-29", guidance, "updated")]]}, } for ticker, payloads in result.items(): for stamp in ("2026-07-01", "2026-09-29"): payloads["guidance@" + stamp] = {"ticker": ticker, "data": [ next(row.copy() for row in payloads["guidance"]["data"] if row["announcementDate"] == stamp)]} return result def write_csv(path, rows): fields = list(dict.fromkeys(key for row in rows for key in row)) with path.open("x", newline="", encoding="utf-8-sig") as file: writer = csv.DictWriter(file, fieldnames=fields) writer.writeheader() # Preserve numbers, but prevent externally supplied text becoming an Excel formula. writer.writerows({key: "'" + value if isinstance(value, str) and value.lstrip().startswith(("=", "+", "-", "@")) else value for key, value in row.items()} for row in rows) def report_html(rows, sample): def display(value): return "—" if value is None else f"{value:.2f}" if isinstance(value, float) else escape(str(value)) cards = [] for row in rows: fields = [("Revenue revision", row.get("revenueRevisionPct"), "%"), ("Target revision", row.get("targetRevisionPct"), "%"), ("Buy-share change", row.get("buyShareChangePp"), " pp"), ("Consensus vs guidance", row.get("consensusVsGuidance"), "")] values = "".join(f"
{label}
{display(value)}{suffix if value is not None else ''}
" for label, value, suffix in fields) width = row.get("score") or 0 cards.append(f'

{escape(row["ticker"])}

Score {display(row.get("score"))}
' f'
{values}
' f'

Decision {row["decisionDate"]} · cutoff {row["cutoffDate"]}
' f'{escape(row.get("rankingStatus", ""))} · {row["rankedCompanyCount"]} eligible companies

') note = "SYNTHETIC EXAMPLE · not reported company values" if sample else "API RESEARCH EXPORT · dated observations" return '' + \ 'Historical analyst stock-ranking inputs
' + \ f'

{note}

Historical analyst stock-ranking inputs

' + \ '

Equal-weight ranks of the selected analyst signals. Guidance is context, not a score penalty. ' + \ 'Scores are relative to the eligible companies at each date, not expected returns. Full inputs, dates and sources: ranking.csv.

' + \ '
' + ''.join(cards) + '
' def export_run(destination, rows, coverage, payloads, metadata): destination.mkdir(parents=True, exist_ok=False) write_csv(destination / "ranking.csv", rows) write_csv(destination / "coverage.csv", coverage) for name, value in [("responses.json", payloads), ("metadata.json", metadata)]: with (destination / name).open("x", encoding="utf-8") as file: json.dump(value, file, indent=2, allow_nan=False) with (destination / "report.html").open("x", encoding="utf-8") as file: file.write(report_html(rows, metadata["syntheticSample"])) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--sample", action="store_true") parser.add_argument("--tickers", nargs="+", default=["AAPL", "MSFT", "PLTR"]) parser.add_argument("--dates", nargs="+", help="Decision dates; each uses information through the previous calendar day") parser.add_argument("--frequency", choices=["annual", "quarterly"], default="annual") parser.add_argument("--period", help="Fixed normalized target label, e.g. 2027 or Q4'26") parser.add_argument("--lookback-days", type=int, default=90) parser.add_argument("--max-age-days", type=int, help="Optional maximum observation gap at either comparison boundary") parser.add_argument("--guidance-basis", help="Exact basis label, only after checking comparability against consensus") parser.add_argument("--signals", nargs="+", choices=SIGNALS, default=list(SIGNALS)) parser.add_argument("--output-dir", default="analyst-ranking") args = parser.parse_args() try: destination = Path(args.output_dir) if destination.exists(): raise ValueError("Choose a new --output-dir; existing exports are never overwritten.") if args.lookback_days <= 0 or (args.max_age_days is not None and args.max_age_days <= 0): raise ValueError("Lookback and optional maximum age must be positive.") if len(set(args.signals)) != len(args.signals): raise ValueError("Do not repeat a ranking signal.") if args.sample: payloads, errors = sample_payloads(), {} dates, frequency, period, basis = ["2026-09-30"], "annual", "2027", "GAAP" if args.dates or args.period or args.frequency != "annual" or args.guidance_basis or args.lookback_days != 90: raise ValueError("The sample uses fixed annual 2027 inputs, a September 30 decision and a 90-day lookback.") else: dates = sorted(set(args.dates or [date.today().isoformat()])) frequency, period, basis = args.frequency, args.period, args.guidance_basis pattern = r"\d{4}" if frequency == "annual" else r"Q[1-4]'\d{2}" if not re.fullmatch(pattern, period or ""): raise ValueError("Set --period to an available normalized period using the chosen --frequency.") tickers = list(dict.fromkeys(t.strip().upper() for t in args.tickers)) if any(not re.fullmatch(r"[A-Z0-9][A-Z0-9.-]{0,19}", ticker) for ticker in tickers): raise ValueError("Invalid stock ticker.") for value in dates: if day(value) > date.today(): raise ValueError("Choose a historical decision date, not a future date.") payloads, errors = {}, {} with make_session() as session: for ticker in tickers: payloads[ticker], errors[ticker] = fetch_histories(session, ticker, frequency, period, dates, args.lookback_days) cutoff = max(day(value) for value in dates) - timedelta(days=1) rows, coverage = [], [] for ticker, histories in payloads.items(): coverage.extend(coverage_rows(ticker, histories, errors.get(ticker, {}), frequency, period, cutoff)) for value in dates: row = build_row(ticker, histories, value, frequency, period, args.lookback_days, args.max_age_days, basis) row["requestNotes"] = "; ".join(f"{key}: {error}" for key, error in errors.get(ticker, {}).items()) rows.append(row) rows = rank_rows(rows, args.signals) metadata = dict(retrievedAtUtc=datetime.now(timezone.utc).isoformat(), syntheticSample=args.sample, decisionDates=dates, tickers=list(payloads), frequency=frequency, normalizedPeriod=period, lookbackDays=args.lookback_days, maxAgeDays=args.max_age_days, guidanceBasis=basis, signals=args.signals, errors=errors, methodology="Previous-calendar-day cutoffs; last eligible observation, no interpolation. " "Equal-weight ranks use only companies with all selected signals; guidance is unscored. " "Estimate collection dates differ from original analyst publication dates; the unified analyst " "consensus history contains calculated rating and price-target aggregates. This export is not an " "immutable original-data-vintage archive. " "Coverage is for the requested series through the latest cutoff; responses.json retains full fetched history.") export_run(destination, rows, coverage, payloads, metadata) print(f"Saved {len(rows)} rows in {destination}. Open report.html; check coverage.csv and requestNotes before using rankings.") except (ValueError, OSError, KeyError, TypeError) as error: parser.exit(1, f"{error}\n") if __name__ == "__main__": main()