Intraday data, retries and scheduled jobs

"""Prices, 1-minute bars and SEC filings over HTTP, as JSON. Usage:     uv run fastapi dev src/market_data/api.py Interactive documentation is at http://127.0.0.1:8000/docs.""" import timefrom collections.abc import AsyncIterator, Callable, Iteratorfrom contextlib import asynccontextmanagerfrom datetime import date, datetimefrom typing import Annotated import psycopgfrom fastapi import Depends, FastAPI, HTTPException, Query, Requestfrom psycopg.rows import dict_rowfrom psycopg_pool import ConnectionPoolfrom pydantic import BaseModel, TypeAdapter from market_data.config import get_settingsfrom market_data.filings import describefrom market_data.intraday import get_minutes PRICES = """SELECT day, open, high, low, close, volumeFROM pricesWHERE symbol = %sORDER BY day DESCLIMIT %s""" FILINGS = """SELECT filed, symbol, form, items, urlFROM filingsORDER BY filed DESC, symbol, accession DESCLIMIT %s"""  class Price(BaseModel):    """One day's bar."""     day: date    open: float    high: float    low: float    close: float    volume: int  class Minute(BaseModel):    """One 1-minute bar."""     time: datetime    open: float    high: float    low: float    close: float    volume: int  class Filing(BaseModel):    """An SEC filing and its main document."""     filed: date    symbol: str    form: str    description: str    url: str  # Checks a list of plain rows against Minute, and builds the models.MINUTES = TypeAdapter(list[Minute])  def latest_minutes(symbol: str) -> list[Minute]:    """The latest trading day's minutes, from Alpaca."""    bars = get_minutes(symbol).reset_index()    return MINUTES.validate_python(bars.to_dict("records"))  class MinutesCache:    """Each symbol's minutes, reloaded at most once per `seconds`."""     def __init__(self, load: Callable[[str], list[Minute]], seconds: float) -> None:        self.load = load        self.seconds = seconds        self.saved: dict[str, tuple[float, list[Minute]]] = {}     def get(self, symbol: str) -> list[Minute]:        loaded_at, minutes = self.saved.get(symbol, (0.0, []))        if time.monotonic() - loaded_at > self.seconds:            minutes = self.load(symbol)            self.saved[symbol] = (time.monotonic(), minutes)        return minutes  @asynccontextmanagerasync def lifespan(app: FastAPI) -> AsyncIterator[None]:    """Open a pool of database connections on startup; close it on shutdown."""    with ConnectionPool(get_settings().database_url) as pool:        app.state.pool = pool        # Alpaca's minutes change once a minute, so each symbol is fetched at most that often.        app.state.minutes = MinutesCache(latest_minutes, seconds=60)        yield  app = FastAPI(title="market-data", lifespan=lifespan)  def get_conn(request: Request) -> Iterator[psycopg.Connection]:    """A pooled connection for the length of one request."""    with request.app.state.pool.connection() as conn:        yield conn  Conn = Annotated[psycopg.Connection, Depends(get_conn)]  @app.get("/symbols")def symbols(conn: Conn) -> list[str]:    """Every symbol with daily prices."""    rows = conn.execute("SELECT DISTINCT symbol FROM prices ORDER BY symbol")    return [symbol for (symbol,) in rows]  @app.get("/prices/{symbol}")def prices(    conn: Conn, symbol: str, days: Annotated[int, Query(ge=1, le=6000)] = 30) -> list[Price]:    """A symbol's latest daily bars, oldest first."""    rows = conn.cursor(row_factory=dict_row).execute(PRICES, (symbol, days))    latest = [Price(**row) for row in rows]    if not latest:        raise HTTPException(status_code=404, detail=f"No prices for {symbol}")    return latest[::-1]  @app.get("/minutes/{symbol}")def minutes(    request: Request, symbol: str, last: Annotated[int | None, Query(ge=1)] = None) -> list[Minute]:    """The latest trading day's 1-minute bars, at most a minute old."""    bars: list[Minute] = request.app.state.minutes.get(symbol)    return bars[-last:] if last else bars  @app.get("/filings")def filings(    conn: Conn, limit: Annotated[int, Query(ge=1, le=100)] = 10) -> list[Filing]:    """The newest filings across the watchlist."""    rows = conn.cursor(row_factory=dict_row).execute(FILINGS, (limit,))    return [        Filing(            filed=row["filed"],            symbol=row["symbol"],            form=row["form"],            description=describe(row["form"], row["items"]),            url=row["url"],        )        for row in rows    ]

from collections.abc import Iteratorfrom datetime import datetimefrom zoneinfo import ZoneInfo import psycopgimport pytestfrom fastapi.testclient import TestClient from market_data import apifrom market_data.config import get_settings PRICES = [    ("AAPL", "2026-10-01", 310.49, 318.71, 309.90, 316.80, 61_000_000),    ("AAPL", "2026-10-02", 318.80, 328.60, 315.21, 326.04, 46_100_000),    ("AAPL", "2026-10-05", 323.68, 328.43, 322.77, 326.80, 42_600_000),    ("SPY", "2026-10-05", 671.10, 674.20, 669.80, 673.50, 51_000_000),]FILINGS = [    ("0000320193-26-000071", "AAPL", "10-Q", "2026-07-31", "2026-06-27", "", "a"),    ("0001045810-26-000201", "NVDA", "8-K", "2026-09-03", "2026-09-03", "8.01", "b"),]  @pytest.fixturedef client(database: str, monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:    """The API on the test database, holding the rows above."""    with psycopg.connect(database) as conn:        conn.execute("TRUNCATE prices, filings")        with conn.cursor() as cur:            cur.executemany(                "INSERT INTO prices VALUES (%s, %s, %s, %s, %s, %s, %s)", PRICES            )            cur.executemany(                "INSERT INTO filings VALUES (%s, %s, %s, %s, %s, %s, %s)", FILINGS            )    monkeypatch.setenv("DATABASE_URL", database)    get_settings.cache_clear()    with TestClient(api.app) as test_client:        yield test_client    get_settings.cache_clear()  def test_symbols(client: TestClient) -> None:    assert client.get("/symbols").json() == ["AAPL", "SPY"]  def test_prices_are_the_latest_days_oldest_first(client: TestClient) -> None:    days = client.get("/prices/AAPL?days=2").json()    assert [d["day"] for d in days] == ["2026-10-02", "2026-10-05"]    assert days[-1]["close"] == 326.80  def test_an_unknown_symbol_is_404(client: TestClient) -> None:    response = client.get("/prices/ZZZZ")    assert response.status_code == 404    assert response.json() == {"detail": "No prices for ZZZZ"}  def test_days_must_be_at_least_1(client: TestClient) -> None:    assert client.get("/prices/AAPL?days=0").status_code == 422  def test_filings_are_newest_first_and_described(client: TestClient) -> None:    filings = client.get("/filings").json()    assert [(f["symbol"], f["description"]) for f in filings] == [        ("NVDA", "Other events"),        ("AAPL", "Quarterly report"),    ]  def test_minutes_are_loaded_at_most_once_a_minute() -> None:    loads: list[str] = []     def load(symbol: str) -> list[api.Minute]:        loads.append(symbol)        return [            api.Minute(                time=datetime(2026, 10, 5, 9, 31, tzinfo=ZoneInfo("America/New_York")),                open=323.68,                high=323.9,                low=323.6,                close=323.7,                volume=100,            )        ]     cache = api.MinutesCache(load, seconds=60)    cache.get("AAPL")    cache.get("AAPL")    assert loads == ["AAPL"]

import { env } from "@/lib/env"; // The browser asks this route for a symbol's minutes, and the route asks the market-data API,// so the API's address stays on the server.export async function GET(  _request: Request,  context: RouteContext<"/api/minutes/[symbol]">,) {  const { symbol } = await context.params;  const response = await fetch(    `${env.MARKET_API_URL}/minutes/${encodeURIComponent(symbol)}`,  );  return new Response(response.body, {    status: response.status,    headers: { "content-type": "application/json" },  });}

"use client"; import { useEffect, useState } from "react";import { PriceChart } from "@/components/price-chart";import { Stat } from "@/components/stat";import { Alert, AlertDescription, AlertTitle } from "@/components/ui/alert";import {  Card,  CardContent,  CardDescription,  CardHeader,  CardTitle,} from "@/components/ui/card";import { Skeleton } from "@/components/ui/skeleton";import type { components } from "@/lib/api";import { percent, price, tone, volume } from "@/lib/format"; type Minute = components["schemas"]["Minute"]; // The API's minutes change once a minute, so the page asks for them again every minute.const REFRESH_MS = 60_000; export function IntradayView({ symbol }: { symbol: string }) {  const [minutes, setMinutes] = useState<Minute[]>([]);  const [failed, setFailed] = useState(false);   useEffect(() => {    let active = true;    async function load() {      try {        const response = await fetch(          `/api/minutes/${encodeURIComponent(symbol)}`,        );        if (!response.ok) {          throw new Error(String(response.status));        }        const bars: Minute[] = await response.json();        if (active) {          setMinutes(bars);          setFailed(false);        }      } catch {        if (active) {          setFailed(true);        }      }    }    load();    const timer = setInterval(load, REFRESH_MS);    return () => {      active = false;      clearInterval(timer);    };  }, [symbol]);   if (minutes.length === 0) {    return failed ? <Unavailable /> : <Skeleton className="h-96" />;  }  const first = minutes[0];  const last = minutes[minutes.length - 1];  const change = last.close / first.open - 1;  const high = Math.max(...minutes.map((m) => m.high));  const low = Math.min(...minutes.map((m) => m.low));  const traded = minutes.reduce((sum, m) => sum + m.volume, 0);  return (    <div className="grid gap-4">      {failed && <Unavailable />}      <div className="grid gap-3 sm:grid-cols-2 xl:grid-cols-4">        <Stat          label="Last"          value={price(last.close)}          note={`${percent(change)} since the open`}          noteClass={tone(change)}        />        <Stat label="High" value={price(high)} />        <Stat label="Low" value={price(low)} />        <Stat label="Volume" value={volume(traded)} />      </div>      <Card>        <CardHeader>          <CardTitle>{symbol} 1-minute closes</CardTitle>          <CardDescription>            {first.time.slice(0, 10)}, {first.time.slice(11, 16)} to{" "}            {last.time.slice(11, 16)} ET          </CardDescription>        </CardHeader>        <CardContent>          <PriceChart            data={minutes.map((m) => ({              label: m.time.slice(11, 16),              close: m.close,            }))}          />        </CardContent>      </Card>    </div>  );} function Unavailable() {  return (    <Alert variant="destructive">      <AlertTitle>Market data unavailable</AlertTitle>      <AlertDescription>        The market-data API isn’t responding. Showing the last data received.      </AlertDescription>    </Alert>  );}

import type { Metadata } from "next";import { IntradayView } from "@/components/intraday-view";import { PageHeader } from "@/components/page-header";import { Badge } from "@/components/ui/badge"; export const metadata: Metadata = { title: "Intraday" }; const SYMBOL = "AAPL"; export default function Intraday() {  return (    <div className="grid gap-4">      <PageHeader        title="Intraday"        description="1-minute bars from Alpaca, 15 minutes behind the market"      >        <Badge variant="outline" className="font-mono">          {SYMBOL}        </Badge>      </PageHeader>      <IntradayView symbol={SYMBOL} />    </div>  );}

import { Activity, ChartLine, House, type LucideIcon } from "lucide-react"; export type NavItem = { title: string; href: string; icon: LucideIcon };export type NavGroup = {  label: string;  description?: string;  items: NavItem[];}; // The site's sections, in the sidebar's order. A new page is added here.export const NAV: NavGroup[] = [  { label: "Overview", items: [{ title: "Home", href: "/", icon: House }] },  {    label: "Markets",    description: "Daily and intraday prices from my market-data API.",    items: [      { title: "Prices", href: "/prices", icon: ChartLine },      { title: "Intraday", href: "/intraday", icon: Activity },    ],  },];

"""GET JSON with retries: transient failures are retried after growing waits, andevery failure is raised as a FetchError.""" import loggingimport randomimport timefrom typing import Any import httpx2 log = logging.getLogger(__name__) ATTEMPTS = 4FIRST_WAIT = 1.0# Statuses worth retrying: rate limiting and server errors.RETRY = {429, 500, 502, 503, 504}  class FetchError(Exception):    """A request that failed: refused, or failing on every attempt."""  def get_json(    client: httpx2.Client, url: str, params: dict[str, Any] | None = None) -> Any:    """GET a URL and its query parameters with the client and return its JSON,    retrying transient failures."""    # The full address, with the base URL and the query, for the log and the error.    target = client.build_request("GET", url, params=params).url    problem = ""    for attempt in range(ATTEMPTS):        if attempt:            wait = FIRST_WAIT * 2 ** (attempt - 1) * random.uniform(0.8, 1.2)            log.warning("%s: %s, retrying in %.1f s", target, problem, wait)            time.sleep(wait)        try:            response = client.get(url, params=params)        except httpx2.TransportError as error:            problem = type(error).__name__            continue        status = f"{response.status_code} {response.reason_phrase}"        if response.status_code in RETRY:            problem = status            continue        if response.is_error:            # Refused, such as 404: asking again would fail the same way.            raise FetchError(f"{target}: {status}")        return response.json()    raise FetchError(f"{target} failed {ATTEMPTS} times, last with {problem}")

"""Alpaca's market data: snapshots, daily bars and 1-minute bars. Snapshots are 15 minutes behind the market and cover every US exchange. The freeplan has bars since 2016, up to 15 minutes ago. The data is for personal use.""" from datetime import date, datetime, time, timedeltafrom functools import lru_cachefrom typing import Anyfrom zoneinfo import ZoneInfo import httpx2 from market_data.config import get_settingsfrom market_data.net import get_json BASE_URL = "https://data.alpaca.markets"# The market's time zone. Alpaca gives every time in UTC.NEW_YORK = ZoneInfo("America/New_York")# The free plan's full-market data ends 15 minutes before now.DELAY = timedelta(minutes=15)  @lru_cachedef client() -> httpx2.Client:    """One client for every request to Alpaca, made the first time it is needed:    it keeps its connection open between requests and sends your keys with each."""    settings = get_settings()    keys = {        "APCA-API-KEY-ID": settings.apca_api_key_id,        "APCA-API-SECRET-KEY": settings.apca_api_secret_key.get_secret_value(),    }    return httpx2.Client(base_url=BASE_URL, headers=keys, timeout=10)  def get(path: str, params: dict[str, Any]) -> Any:    """GET a path with its query and return the JSON, retrying transient failures."""    return get_json(client(), path, params)  def snapshots(symbols: list[str]) -> dict[str, Any]:    """Each symbol's latest trade, quote and daily bars, 15 minutes behind."""    query = {"symbols": ",".join(symbols), "feed": "delayed_sip"}    found: dict[str, Any] = get("/v2/stocks/snapshots", query)    return found  def bars(symbol: str, query: dict[str, Any]) -> list[dict[str, Any]]:    """A symbol's bars, oldest first, following Alpaca's pages to the last."""    query = {"symbols": symbol, "feed": "sip", "limit": 10000, **query}    found: list[dict[str, Any]] = []    while True:        page = get("/v2/stocks/bars", query)        found += page["bars"].get(symbol, [])        if not page["next_page_token"]:            return found        query["page_token"] = page["next_page_token"]  def day_of(timestamp: str) -> date:    """The New York date of one of Alpaca's UTC times."""    return datetime.fromisoformat(timestamp).astimezone(NEW_YORK).date()  def daily_bars(symbol: str) -> list[dict[str, Any]]:    """Every trading day's bar since 2016, adjusted for splits."""    query = {"timeframe": "1Day", "start": "2016-01-01", "adjustment": "split"}    return bars(symbol, query)  def latest_day(symbol: str) -> date | None:    """The latest day the symbol traded: today once it has, or the last day it did.    None if Alpaca has no such symbol."""    snapshot = snapshots([symbol]).get(symbol)    return day_of(snapshot["dailyBar"]["t"]) if snapshot else None  def minute_bars(symbol: str, day: date) -> list[dict[str, Any]]:    """A day's 1-minute bars, from 09:30 New York time to the last minute before    16:00, or to 15 minutes ago if that is sooner. Each is labelled by its start."""    opening = datetime.combine(day, time(9, 30), NEW_YORK)    last = datetime.combine(day, time(15, 59), NEW_YORK)    end = min(last, datetime.now(NEW_YORK) - DELAY)    # A day still to come has no minutes yet, and Alpaca refuses to be asked for them.    if end < opening:        return []    query = {"timeframe": "1Min", "start": opening.isoformat(), "end": end.isoformat()}    return bars(symbol, query)

"""SEC EDGAR: company identifiers and recent filings. The SEC requires every request to name who is making it, in the User-Agent header.""" from functools import lru_cachefrom typing import Any import httpx2 from market_data.config import get_settingsfrom market_data.net import get_json TICKERS_URL = "https://www.sec.gov/files/company_tickers.json"SUBMISSIONS_URL = "https://data.sec.gov/submissions/CIK{cik:010d}.json"DOCUMENT_URL = "https://www.sec.gov/Archives/edgar/data/{cik}/{folder}/{document}"  @lru_cachedef client() -> httpx2.Client:    """A client that sends the User-Agent the SEC requires."""    headers = {"User-Agent": get_settings().sec_user_agent}    return httpx2.Client(headers=headers, follow_redirects=True, timeout=10)  def get(url: str) -> Any:    """GET a URL from the SEC and return its JSON, retrying transient failures."""    return get_json(client(), url)  def ciks() -> dict[str, int]:    """Each listed company's CIK, its SEC identifier, by ticker."""    rows: dict[str, dict[str, Any]] = get(TICKERS_URL)    return {row["ticker"]: row["cik_str"] for row in rows.values()}  def recent_filings(cik: int) -> dict[str, list[Any]]:    """A company's recent filings, as columns of equal length."""    submissions = get(SUBMISSIONS_URL.format(cik=cik))    recent: dict[str, list[Any]] = submissions["filings"]["recent"]    return recent  def document_url(cik: int, accession: str, document: str) -> str:    """The address of a filing's main document."""    folder = accession.replace("-", "")    return DOCUMENT_URL.format(cik=cik, folder=folder, document=document)

"""What every command sets up before it runs.""" import logging  def setup_logging() -> None:    """Print warnings and errors as 'LEVEL module: message'."""    logging.basicConfig(format="%(levelname)s %(name)s: %(message)s")

"""Print a symbol's latest quote: last price, bid, ask and spread. Usage:     uv run quote AAPL""" import argparseimport sysfrom typing import Any from market_data import alpacafrom market_data.cli import setup_logging  def describe(symbol: str, snapshot: dict[str, Any]) -> str:    """A quote on one line: the latest trade's price, the bid, the ask and the spread."""    last = snapshot["latestTrade"]["p"]    bid = snapshot["latestQuote"]["bp"]    ask = snapshot["latestQuote"]["ap"]    return (        f"{symbol}  last {last:.2f}  bid {bid:.2f}  ask {ask:.2f}  "        f"spread {ask - bid:.2f}"    )  def main() -> None:    setup_logging()    parser = argparse.ArgumentParser(description="Print a symbol's latest quote.")    parser.add_argument(        "symbol", nargs="?", default="AAPL", help="ticker symbol (default: AAPL)"    )    symbol = parser.parse_args().symbol    snapshots = alpaca.snapshots([symbol])    if symbol not in snapshots:        sys.exit(f"No quote for {symbol}: check the ticker.")    print(describe(symbol, snapshots[symbol]))  if __name__ == "__main__":    main()

import httpx2import pytest from market_data import net URL = "https://example.com/data.json"  def server(*statuses: int) -> tuple[httpx2.Client, list[str]]:    """A client whose server answers with these statuses in turn, then 200."""    calls: list[str] = []     def answer(request: httpx2.Request) -> httpx2.Response:        calls.append(str(request.url))        status = statuses[len(calls) - 1] if len(calls) <= len(statuses) else 200        return httpx2.Response(status, json={"ok": True})     return httpx2.Client(transport=httpx2.MockTransport(answer)), calls  @pytest.fixturedef waits(monkeypatch: pytest.MonkeyPatch) -> list[float]:    """Every wait get_json asks for, recorded instead of slept."""    asked: list[float] = []    monkeypatch.setattr(net.time, "sleep", asked.append)    return asked  def test_a_transient_failure_is_retried(waits: list[float]) -> None:    client, calls = server(503, 503)    assert net.get_json(client, URL) == {"ok": True}    assert len(calls) == 3  def test_each_wait_is_longer(waits: list[float]) -> None:    client, _ = server(503, 503, 503)    net.get_json(client, URL)    assert waits[0] < waits[1] < waits[2]  def test_it_gives_up_after_the_last_attempt(waits: list[float]) -> None:    client, calls = server(503, 503, 503, 503)    with pytest.raises(net.FetchError, match="failed 4 times"):        net.get_json(client, URL)    assert len(calls) == 4  def test_a_client_error_is_not_retried(waits: list[float]) -> None:    client, calls = server(404)    with pytest.raises(net.FetchError, match="404 Not Found"):        net.get_json(client, URL)    assert len(calls) == 1  def test_a_lost_connection_is_retried(waits: list[float]) -> None:    attempts: list[int] = []     def answer(request: httpx2.Request) -> httpx2.Response:        attempts.append(1)        if len(attempts) == 1:            raise httpx2.ConnectError("connection refused", request=request)        return httpx2.Response(200, json={"ok": True})     client = httpx2.Client(transport=httpx2.MockTransport(answer))    assert net.get_json(client, URL) == {"ok": True}    assert len(waits) == 1

CREATE TABLE minutes (    symbol text NOT NULL,    time timestamptz NOT NULL,    open numeric(12, 4) NOT NULL,    high numeric(12, 4) NOT NULL,    low numeric(12, 4) NOT NULL,    close numeric(12, 4) NOT NULL,    volume bigint NOT NULL,    PRIMARY KEY (symbol, time));

"""Load a trading day's 1-minute bars into the database. Schedule it after every close, so the database keeps up by itself. Usage:     uv run minutes    uv run minutes AAPL --day 2026-10-05 With no symbols, it loads the watchlist; with no day, the latest each traded.""" import argparseimport loggingimport sysfrom datetime import date import psycopg from market_data.cli import setup_loggingfrom market_data.db import connectfrom market_data.intraday import get_minutesfrom market_data.net import FetchErrorfrom market_data.watchlist import SYMBOLS log = logging.getLogger(__name__) UPSERT = """INSERT INTO minutes (symbol, time, open, high, low, close, volume)VALUES (%s, %s, %s, %s, %s, %s, %s)ON CONFLICT (symbol, time) DO UPDATE SET    open = EXCLUDED.open,    high = EXCLUDED.high,    low = EXCLUDED.low,    close = EXCLUDED.close,    volume = EXCLUDED.volume""" DAYS = """SELECT symbol, (time AT TIME ZONE 'America/New_York')::date AS day, count(*)FROM minutesGROUP BY symbol, dayORDER BY symbol, day"""  def record(conn: psycopg.Connection, symbol: str, day: date | None) -> str:    """Upsert a symbol's day of minutes; return the day and the count."""    minutes = get_minutes(symbol, day).reset_index().assign(symbol=symbol)    if minutes.empty:        return "no minutes that day: check the ticker and the day"    rows = minutes[["symbol", "time", "open", "high", "low", "close", "volume"]]    with conn.cursor() as cur:        cur.executemany(UPSERT, rows.itertuples(index=False, name=None))    return f"{len(minutes)} minutes on {minutes['time'].iloc[0]:%d %b %Y}"  def main() -> None:    setup_logging()    parser = argparse.ArgumentParser(description="Load a day's 1-minute bars.")    parser.add_argument(        "symbols",        nargs="*",        default=SYMBOLS,        help="ticker symbols (default: the watchlist)",    )    parser.add_argument(        "--day", type=date.fromisoformat, help="YYYY-MM-DD (default: the latest)"    )    args = parser.parse_args()    failed = []    with connect() as conn:        for symbol in args.symbols:            try:                print(f"{symbol}: {record(conn, symbol, args.day)}")            except FetchError as error:                log.error("%s: %s", symbol, error)                failed.append(symbol)        print("\nsymbol  day         minutes")        for symbol, day, count in conn.execute(DAYS):            print(f"{symbol:<8}{day}  {count:>7}")    if failed:        sys.exit(1)  if __name__ == "__main__":    main()

[project.scripts]quote = "market_data.quote:main"watchlist = "market_data.watchlist:main"history = "market_data.history:main"returns = "market_data.returns:main"adjust = "market_data.adjust:main"intraday = "market_data.intraday:main"migrate = "market_data.db:main"recorder = "market_data.recorder:main"report = "market_data.report:main"filings = "market_data.filings:main"minutes = "market_data.minutes:main"

# What the scheduled recorder printsminutes.log

# market-data US equity market data in Python: quotes, daily bars since 2016, returns, split anddividend adjustment, a trading day's 1-minute bars, and SEC filings. Daily bars, filingsand each day's 1-minute bars are stored in PostgreSQL, and an API serves prices,1-minute bars and filings. Prices come from Alpaca's market data API, free with an Alpaca account: quotes from everyUS exchange, 15 minutes behind the market. The data is for personal use, so thisrepository holds no prices: each command downloads its own. ## Setup Install [uv](https://docs.astral.sh/uv/) and PostgreSQL, and make API keys in an[Alpaca](https://alpaca.markets/) paper trading account. Copy `.env.example` to `.env` andset your own values, then run: ```uv syncuv run migrate``` ## Commands | Command | Description || --- | --- || `uv run quote AAPL` | Latest quote: last price, bid, ask and spread || `uv run watchlist` | Watchlist quotes, refreshed every minute || `uv run history AAPL` | Daily bars since 2016, saved to `data/` and charted || `uv run returns AAPL` | Best and worst days, total return, compound annual growth, yearly returns || `uv run adjust` | Split detection and adjustment, and total return with dividends || `uv run intraday AAPL` | A trading day's 1-minute bars: day bar, VWAP and volume by hour || `uv run migrate` | Apply new database migrations || `uv run recorder` | Load the watchlist's daily bars into the database || `uv run report` | Latest closes and the five best days, from the database || `uv run filings` | Load the watchlist's 10-K, 10-Q and 8-K filings, and print the latest || `uv run minutes` | Load a trading day's 1-minute bars; schedule it after each close | Each command takes `--help`. ## API ```uv run fastapi dev src/market_data/api.py``` Interactive documentation is at http://127.0.0.1:8000/docs. ## Development ```uv run ruff formatuv run ruff checkuv run mypyuv run pytest``` The API's tests need PostgreSQL at `DATABASE_URL`. They create a `market_data_test`database there and drop it when they finish.