Saving your work and your data

# market-data US equity market data in Python: quotes, daily bars since 2016, returns, split anddividend adjustment, and a trading day's 1-minute bars. 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 make API keys in an[Alpaca](https://alpaca.markets/) paper trading account. Copy `.env.example` to `.env` andset your keys, then run: ```uv sync``` Commands that download prices read your keys from `.env` through `--env-file`. ## Commands | Command | Description || --- | --- || `uv run --env-file .env quote AAPL` | Latest quote: last price, bid, ask and spread || `uv run --env-file .env watchlist` | Watchlist quotes, refreshed every minute || `uv run --env-file .env 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 --env-file .env intraday AAPL` | A trading day's 1-minute bars: day bar, VWAP and volume by hour | Each command takes `--help`. ## Development ```uv run ruff formatuv run ruff checkuv run mypyuv run pytest```

[tool.mypy]files = ["src"]strict = trueuntyped_calls_exclude = ["matplotlib"]

# Check every push: formatting, lint, types and tests.name: Check on: [push, pull_request] jobs:  check:    runs-on: ubuntu-latest    steps:      - uses: actions/checkout@v7      - uses: astral-sh/setup-uv@v10.2.0      - run: uv run ruff format --check      - run: uv run ruff check      - run: uv run mypy      - run: uv run pytest

APCA_API_KEY_ID=your-key-idAPCA_API_SECRET_KEY=your-secret-keyDATABASE_URL=postgresql://postgres:choose-a-password@localhost:5432/market

# Copy to .env and set your own values. .env is in .gitignore; this file is not.APCA_API_KEY_ID=your-key-idAPCA_API_SECRET_KEY=your-secret-keyDATABASE_URL=postgresql://postgres:your-password@localhost:5432/market

"""Settings that differ between computers, read from the environment or .env. Every field without a default must be set, or the first command that needs a settingstops with a message naming the missing ones. .env.example lists them.""" from functools import lru_cache from pydantic import SecretStrfrom pydantic_settings import BaseSettings, SettingsConfigDict  class Settings(BaseSettings):    model_config = SettingsConfigDict(env_file=".env")     apca_api_key_id: str    # A secret: printing the settings shows it as **********.    apca_api_secret_key: SecretStr    database_url: str  @lru_cachedef get_settings() -> Settings:    """The settings, read once."""    return Settings()

"""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_settings 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."""    response = client().get(path, params=params)    response.raise_for_status()    return response.json()  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)

[tool.mypy]files = ["src"]strict = trueplugins = ["pydantic.mypy"]untyped_calls_exclude = ["matplotlib"]

"""The PostgreSQL database: connections, and migrations that keep its tables up to date. Usage:     uv run migrate""" from importlib.resources import files import psycopg from market_data.config import get_settings MIGRATIONS = files("market_data") / "migrations" APPLIED = """CREATE TABLE IF NOT EXISTS schema_migrations (    version text PRIMARY KEY,    applied_at timestamptz NOT NULL DEFAULT now())"""  def connect() -> psycopg.Connection:    """A connection to the database at DATABASE_URL."""    return psycopg.connect(get_settings().database_url)  def migrate(conn: psycopg.Connection) -> list[str]:    """Apply each migration not yet applied, in order; return the ones applied."""    conn.execute(APPLIED)    done = {        version for (version,) in conn.execute("SELECT version FROM schema_migrations")    }    applied = []    for script in sorted(MIGRATIONS.iterdir(), key=lambda path: path.name):        version = script.name.removesuffix(".sql")        if script.name.endswith(".sql") and version not in done:            conn.execute(script.read_text())            conn.execute(                "INSERT INTO schema_migrations (version) VALUES (%s)", (version,)            )            applied.append(version)    return applied  def main() -> None:    with connect() as conn:        applied = migrate(conn)    print("\n".join(f"applied {version}" for version in applied) or "up to date")  if __name__ == "__main__":    main()

CREATE TABLE prices (    symbol text NOT NULL,    day date 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, day));

[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"

"""Load daily bars into the database, then summarise what it holds. Usage:     uv run recorder AAPL SPY With no symbols, it loads the watchlist.""" import argparse import psycopg from market_data.db import connectfrom market_data.history import get_historyfrom market_data.watchlist import SYMBOLS UPSERT = """INSERT INTO prices (symbol, day, open, high, low, close, volume)VALUES (%s, %s, %s, %s, %s, %s, %s)ON CONFLICT (symbol, day) DO UPDATE SET    open = EXCLUDED.open,    high = EXCLUDED.high,    low = EXCLUDED.low,    close = EXCLUDED.close,    volume = EXCLUDED.volume""" SUMMARY = """SELECT symbol, count(*), min(day), max(day)FROM pricesGROUP BY symbolORDER BY symbol"""  def record(conn: psycopg.Connection, symbol: str) -> int:    """Upsert a symbol's full history; return the number of days."""    prices = get_history(symbol).reset_index().assign(symbol=symbol)    rows = prices[["symbol", "date", "open", "high", "low", "close", "volume"]]    with conn.cursor() as cur:        cur.executemany(UPSERT, rows.itertuples(index=False, name=None))    return len(prices)  def main() -> None:    parser = argparse.ArgumentParser(description="Load daily bars into the database.")    parser.add_argument(        "symbols",        nargs="*",        default=SYMBOLS,        help="ticker symbols (default: the watchlist)",    )    args = parser.parse_args()    with connect() as conn:        for symbol in args.symbols:            print(f"{symbol}: {record(conn, symbol):,} days saved")        print("\nsymbol      days  first       last")        for symbol, days, first, last in conn.execute(SUMMARY):            print(f"{symbol:<6}{days:>10,}  {first}  {last}")  if __name__ == "__main__":    main()

"""Questions for the price database, answered in SQL. Usage:     uv run report""" from market_data.db import connect LATEST = """SELECT symbol, day, closeFROM pricesWHERE day = (SELECT max(day) FROM prices)ORDER BY symbol""" BEST_DAYS = """SELECT symbol, day, close / lag(close) OVER (PARTITION BY symbol ORDER BY day) - 1 AS rFROM pricesORDER BY r DESC NULLS LASTLIMIT 5"""  def main() -> None:    with connect() as conn:        print("Latest close")        for symbol, day, close in conn.execute(LATEST):            print(f"  {symbol:<6}{day}  {close:>10.2f}")        print("\nFive best days")        for symbol, day, r in conn.execute(BEST_DAYS):            print(f"  {symbol:<6}{day}  {r:>+8.2%}")  if __name__ == "__main__":    main()

APCA_API_KEY_ID=your-key-idAPCA_API_SECRET_KEY=your-secret-keyDATABASE_URL=postgresql://postgres:choose-a-password@localhost:5432/marketSEC_USER_AGENT=Your Name [email protected]

# Copy to .env and set your own values. .env is in .gitignore; this file is not.APCA_API_KEY_ID=your-key-idAPCA_API_SECRET_KEY=your-secret-keyDATABASE_URL=postgresql://postgres:your-password@localhost:5432/marketSEC_USER_AGENT=Your Name [email protected]

"""Settings that differ between computers, read from the environment or .env. Every field without a default must be set, or the first command that needs a settingstops with a message naming the missing ones. .env.example lists them.""" from functools import lru_cache from pydantic import SecretStrfrom pydantic_settings import BaseSettings, SettingsConfigDict  class Settings(BaseSettings):    model_config = SettingsConfigDict(env_file=".env")     apca_api_key_id: str    # A secret: printing the settings shows it as **********.    apca_api_secret_key: SecretStr    database_url: str    sec_user_agent: str  @lru_cachedef get_settings() -> Settings:    """The settings, read once."""    return Settings()

"""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_settings 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."""    response = client().get(url)    response.raise_for_status()    return response.json()  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)

"""The watchlist, and a table of its quotes refreshed every minute. Usage:     uv run watchlist Press Ctrl+C to stop.""" import argparseimport timefrom datetime import datetimefrom typing import Any from market_data import alpaca # Every command that works on the watchlist reads it from here.SYMBOLS = ["AAPL", "MSFT", "NVDA", "SPY", "QQQ"]# The symbols that file reports with the SEC. The funds don't.COMPANIES = ["AAPL", "MSFT", "NVDA"]  def row(symbol: str, snapshot: dict[str, Any]) -> str:    """One row: symbol, last price, and change since the previous close."""    last = snapshot["latestTrade"]["p"]    previous = snapshot["prevDailyBar"]["c"]    change = last - previous    return f"{symbol:<6} {last:>10.2f} {change:>+8.2f} {change / previous:>+8.2%}"  def show(symbols: list[str]) -> None:    """The New York time, a header, and a row for each symbol, from one request."""    snapshots = alpaca.snapshots(symbols)    print(f"\nQuotes at {datetime.now(alpaca.NEW_YORK):%H:%M:%S} New York time")    print(f"{'symbol':<6} {'last':>10} {'change':>8} {'%':>8}")    for symbol in symbols:        print(row(symbol, snapshots[symbol]))  def main() -> None:    parser = argparse.ArgumentParser(        description="Show the watchlist's quotes, refreshed every minute."    )    parser.add_argument(        "--every", type=int, default=60, help="seconds between refreshes (default: 60)"    )    args = parser.parse_args()    try:        while True:            show(SYMBOLS)            time.sleep(args.every)    except KeyboardInterrupt:        print("\nStopped.")  if __name__ == "__main__":    main()

"""Load the watchlist's SEC filings into the database, and print the latest. Usage:     uv run filings""" from market_data import secfrom market_data.watchlist import COMPANIES  def main() -> None:    numbers = sec.ciks()    print(f"{len(numbers):,} companies")    for symbol in COMPANIES:        print(f"{symbol:<6}CIK {numbers[symbol]:010d}")  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"

"""Load the watchlist's SEC filings into the database, and print the latest. Usage:     uv run filings""" import pandas as pd from market_data import sec FORMS = ["10-K", "10-Q", "8-K"]  def reports(cik: int) -> pd.DataFrame:    """A company's recent 10-K, 10-Q and 8-K filings, newest first."""    filings = pd.DataFrame(sec.recent_filings(cik))    filings = filings[filings["form"].isin(FORMS)].copy()    filings["url"] = [        sec.document_url(cik, accession, document)        for accession, document in zip(            filings["accessionNumber"], filings["primaryDocument"]        )    ]    columns = ["accessionNumber", "form", "filingDate", "reportDate", "items", "url"]    return filings[columns]  def main() -> None:    cik = sec.ciks()["AAPL"]    apple = reports(cik)    print(f"AAPL, CIK {cik:010d}: {len(apple)} filings\n")    print(apple.head(8).drop(columns="url").to_string(index=False))    print(f"\n{apple['url'].iloc[0]}")  if __name__ == "__main__":    main()

CREATE TABLE filings (    accession text PRIMARY KEY,    symbol text NOT NULL,    form text NOT NULL,    filed date NOT NULL,    period date NOT NULL,    items text NOT NULL,    url text NOT NULL);

"""Load the watchlist's SEC filings into the database, and print the latest. Usage:     uv run filings""" import pandas as pdimport psycopg from market_data import secfrom market_data.db import connectfrom market_data.watchlist import COMPANIES FORMS = ["10-K", "10-Q", "8-K"] # The SEC's names for the 8-K items this feed describes.ITEMS = {    "1.01": "Material agreement",    "2.02": "Results of operations",    "5.02": "Officer or director change",    "5.07": "Shareholder vote",    "7.01": "Regulation FD disclosure",    "8.01": "Other events",} UPSERT = """INSERT INTO filings (accession, symbol, form, filed, period, items, url)VALUES (%s, %s, %s, %s, %s, %s, %s)ON CONFLICT (accession) DO NOTHING""" FEED = """SELECT filed, symbol, form, itemsFROM filingsORDER BY filed DESC, symbol, accession DESCLIMIT 10"""  def reports(cik: int) -> pd.DataFrame:    """A company's recent 10-K, 10-Q and 8-K filings, newest first."""    filings = pd.DataFrame(sec.recent_filings(cik))    filings = filings[filings["form"].isin(FORMS)].copy()    filings["url"] = [        sec.document_url(cik, accession, document)        for accession, document in zip(            filings["accessionNumber"], filings["primaryDocument"]        )    ]    columns = ["accessionNumber", "form", "filingDate", "reportDate", "items", "url"]    return filings[columns]  def describe(form: str, items: str) -> str:    """A filing's type, or for an 8-K, what its items report."""    if form == "10-K":        return "Annual report"    if form == "10-Q":        return "Quarterly report"    reported = [ITEMS[item] for item in items.split(",") if item in ITEMS]    return ", ".join(reported) or "Current report"  def save(conn: psycopg.Connection, symbol: str, cik: int) -> int:    """Insert a company's filings, skipping saved ones; return how many it has."""    filings = reports(cik).assign(symbol=symbol)    columns = [        "accessionNumber",        "symbol",        "form",        "filingDate",        "reportDate",        "items",        "url",    ]    with conn.cursor() as cur:        cur.executemany(UPSERT, filings[columns].itertuples(index=False, name=None))    return len(filings)  def main() -> None:    numbers = sec.ciks()    with connect() as conn:        for symbol in COMPANIES:            print(f"{symbol}: {save(conn, symbol, numbers[symbol])} filings")        print("\nLatest filings")        for filed, symbol, form, items in conn.execute(FEED):            print(f"  {filed}  {symbol:<5} {form:<5} {describe(form, items)}")  if __name__ == "__main__":    main()

# 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 andfilings are stored in PostgreSQL. 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 | Each command takes `--help`. ## Development ```uv run ruff formatuv run ruff checkuv run mypyuv run pytest```