Intraday prices, tests and a package

"""A watchlist: the latest quotes for a few symbols, refreshed every minute. Run it, and press Ctrl+C to stop:     uv run --env-file .env watchlist.py""" import timefrom datetime import datetime from market_data import alpaca SYMBOLS = ["AAPL", "MSFT", "NVDA", "SPY", "QQQ"]REFRESH_SECONDS = 60  def row(symbol, snapshot):    """One table 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):    """Print the time in New York, a header, then one row per symbol."""    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]))  if __name__ == "__main__":    try:        while True:            show(SYMBOLS)            time.sleep(REFRESH_SECONDS)    except KeyboardInterrupt:        print("\nStopped.")

"""Download a share's daily prices since 2016, save them, and chart the close. Run it with a ticker, such as:     uv run --env-file .env history.py AAPL""" import sysfrom pathlib import Path import matplotlib.pyplot as pltimport pandas as pdfrom matplotlib import ticker from market_data import alpaca # Alpaca's short names for a bar's fields, and the names this project uses.COLUMNS = {"o": "open", "h": "high", "l": "low", "c": "close", "v": "volume"}  def get_history(symbol):    """Every trading day's open, high, low, close and volume since 2016, adjusted    for splits, oldest first, dated."""    bars = alpaca.daily_bars(symbol)    prices = pd.DataFrame(bars, columns=list(COLUMNS)).rename(columns=COLUMNS)    prices.index = pd.DatetimeIndex(        [alpaca.day_of(bar["t"]) for bar in bars], name="date"    )    return prices  def chart(prices, symbol, path):    """The close on every day, on a log scale, saved as a picture."""    fig, ax = plt.subplots(figsize=(10, 5))    ax.plot(prices.index, prices["close"], linewidth=1)    ax.set_yscale("log")    ax.yaxis.set_major_locator(ticker.LogLocator(subs=[1, 2, 5]))    ax.yaxis.set_major_formatter(ticker.StrMethodFormatter("{x:,g}"))    ax.yaxis.set_minor_formatter(ticker.NullFormatter())    ax.grid(alpha=0.3)    ax.set_title(f"{symbol} daily close")    ax.set_ylabel("USD, log scale")    fig.savefig(path, dpi=120, bbox_inches="tight")  if __name__ == "__main__":    symbol = sys.argv[1] if len(sys.argv) > 1 else "AAPL"    prices = get_history(symbol)    if prices.empty:        sys.exit(f"No prices for {symbol}: check the ticker.")    print(prices.tail())    first, last = prices.index[0], prices.index[-1]    print(f"\n{len(prices):,} days, from {first:%d %b %Y} to {last:%d %b %Y}")    high = prices["close"].idxmax()    print(f"highest close: {prices['close'].max():.2f} on {high:%d %b %Y}")    Path("data").mkdir(exist_ok=True)    Path("charts").mkdir(exist_ok=True)    prices.to_csv(f"data/{symbol}.csv")    chart(prices, symbol, f"charts/{symbol}.png")    print(f"saved data/{symbol}.csv and charts/{symbol}.png")

"""A trading day's 1-minute bars: the day's bar, its VWAP and volume by hour. Run it with a ticker, such as:     uv run --env-file .env intraday.py AAPL""" import sysfrom pathlib import Path import matplotlib.dates as mdatesimport matplotlib.pyplot as pltimport pandas as pdfrom matplotlib import ticker from market_data import alpaca # Alpaca's short names for a bar's fields, and the names this project uses.COLUMNS = {    "o": "open",    "h": "high",    "l": "low",    "c": "close",    "v": "volume",    "vw": "vwap",}  def get_minutes(symbol, day=None):    """A day's 1-minute bars, oldest first, labelled by their start in New York    time: the latest day the symbol traded, if no day is given. Empty if Alpaca has    no such symbol, or the market wasn't open that day."""    day = day or alpaca.latest_day(symbol)    bars = alpaca.minute_bars(symbol, day) if day else []    minutes = pd.DataFrame(bars, columns=list(COLUMNS)).rename(columns=COLUMNS)    times = pd.to_datetime([bar["t"] for bar in bars], utc=True)    minutes.index = times.tz_convert(alpaca.NEW_YORK).rename("time")    return minutes  def day_bar(minutes):    """The whole day as one bar, built from its minutes."""    return {        "open": minutes["open"].iloc[0],        "high": minutes["high"].max(),        "low": minutes["low"].min(),        "close": minutes["close"].iloc[-1],        "volume": minutes["volume"].sum(),    }  def vwap(minutes):    """The day's volume-weighted average price, from each minute's own."""    traded = (minutes["vwap"] * minutes["volume"]).sum()    return traded / minutes["volume"].sum()  def volume_by_hour(minutes):    """The shares traded in each hour on the clock: 9 is the first half hour."""    return minutes["volume"].groupby(minutes.index.hour).sum()  def chart(minutes, symbol, path):    """The price and its VWAP above, each minute's volume below, saved as a picture."""    fig, (top, bottom) = plt.subplots(        2, 1, figsize=(10, 6), sharex=True, height_ratios=[3, 1]    )    top.plot(minutes.index, minutes["close"], linewidth=1, label="price")    top.axhline(vwap(minutes), color="tab:orange", linestyle="--", label="VWAP")    top.set_title(f"{symbol} 1-minute bars, {minutes.index[0]:%d %b %Y}")    top.set_ylabel("USD")    top.legend()    bottom.bar(minutes.index, minutes["volume"], width=1 / (24 * 60), color="gray")    bottom.set_ylabel("shares")    bottom.yaxis.set_major_formatter(ticker.StrMethodFormatter("{x:,.0f}"))    bottom.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M", tz=alpaca.NEW_YORK))    fig.savefig(path, dpi=120, bbox_inches="tight")  if __name__ == "__main__":    symbol = sys.argv[1] if len(sys.argv) > 1 else "AAPL"    minutes = get_minutes(symbol)    if minutes.empty:        sys.exit(f"No minutes for {symbol}: check the ticker.")    first, last = minutes.index[0], minutes.index[-1]    print(        f"{len(minutes)} minutes on {first:%d %b %Y}, {first:%H:%M} to {last:%H:%M}\n"    )    bar = day_bar(minutes)    for field in ("open", "high", "low", "close"):        print(f"{field:<8}{bar[field]:.2f}")    print(f"{'volume':<8}{bar['volume']:,}")    print(f"{'VWAP':<8}{vwap(minutes):.2f}")    hours = volume_by_hour(minutes)    print("\nhour      shares")    for hour, shares in hours.items():        print(f"{hour:>4}{shares:>12,}  {'#' * round(30 * shares / hours.max())}")    Path("data").mkdir(exist_ok=True)    Path("charts").mkdir(exist_ok=True)    day = f"{symbol}-{first:%Y-%m-%d}"    minutes.to_csv(f"data/{day}.csv")    chart(minutes, symbol, f"charts/{day}.png")    print(f"\nsaved data/{day}.csv and charts/{day}.png")

import pandas as pdimport pytest from market_data.returns import (    calendar_years,    daily_returns,    total_return,    yearly_growth,)  def test_daily_returns():    close = pd.Series([100.0, 110.0, 99.0])    assert daily_returns(close).tolist() == pytest.approx([0.10, -0.10])  def test_total_return_compounds():    close = pd.Series([100.0, 110.0, 99.0])    assert total_return(close) == pytest.approx(-0.01)  def test_yearly_growth_compounds_back_to_the_total():    dates = pd.to_datetime(["2020-01-01", "2022-01-01"])    close = pd.Series([100.0, 200.0], index=dates)    years = 731 / 365.25  # 2020 had 366 days, 2021 had 365    assert (1 + yearly_growth(close)) ** years == pytest.approx(2.0)  def test_calendar_years_start_from_the_year_before():    dates = pd.to_datetime(["2024-06-03", "2024-12-31", "2025-06-02", "2025-12-31"])    close = pd.Series([90.0, 100.0, 130.0, 120.0], index=dates)    years = calendar_years(close)    assert years.index.year.tolist() == [2025]    assert years.tolist() == pytest.approx([0.20])

import pandas as pdimport pytest from market_data.adjust import adjust, find_splits, total_returns DAYS = pd.to_datetime(["2026-03-02", "2026-03-03", "2026-03-04"])  def raw_prices():    """Three days of a share that split 2 for 1 on the second day."""    return pd.DataFrame(        {"close": [100.0, 51.0, 52.0], "volume": [1000, 2200, 1800], "dividend": 0.0},        index=DAYS,    )  def test_find_splits_finds_a_2_for_1():    assert find_splits(raw_prices()["close"]) == {DAYS[1]: 2}  def test_find_splits_ignores_an_ordinary_fall():    close = pd.Series([100.0, 90.0, 95.0], index=DAYS)    assert find_splits(close) == {}  def test_adjust_divides_prices_and_multiplies_volume_before_the_split():    adjusted = adjust(raw_prices(), {DAYS[1]: 2})    assert adjusted["close"].tolist() == [50.0, 51.0, 52.0]    assert adjusted["volume"].tolist() == [2000, 2200, 1800]  def test_adjust_leaves_the_raw_prices_alone():    raw = raw_prices()    adjust(raw, {DAYS[1]: 2})    assert raw["close"].tolist() == [100.0, 51.0, 52.0]  def test_total_returns_count_the_dividend():    close = pd.Series([50.0, 49.75], index=DAYS[:2])    dividend = pd.Series([0.0, 0.5], index=DAYS[:2])    assert total_returns(close, dividend).tolist() == pytest.approx([0.005])

import pandas as pdimport pytest from market_data.intraday import day_bar, volume_by_hour, vwap TIMES = pd.to_datetime(["2026-10-06 09:30", "2026-10-06 09:31", "2026-10-06 10:00"])  def three_minutes():    """Three minutes, two in the first half hour and one after 10:00."""    return pd.DataFrame(        {            "open": [10.0, 11.0, 12.0],            "high": [11.0, 12.0, 13.0],            "low": [9.0, 10.0, 11.0],            "close": [10.0, 11.0, 12.0],            "volume": [100, 100, 200],            "vwap": [10.2, 11.1, 12.0],        },        index=TIMES,    )  def test_day_bar_is_built_from_its_minutes():    bar = day_bar(three_minutes())    assert bar == {"open": 10.0, "high": 13.0, "low": 9.0, "close": 12.0, "volume": 400}  def test_vwap_weights_each_minute_by_its_volume():    # (10.2 × 100 + 11.1 × 100 + 12 × 200) ÷ 400 = 4530 ÷ 400 = 11.325.    assert vwap(three_minutes()) == pytest.approx(11.325)  def test_volume_by_hour_adds_up_each_hour():    hours = volume_by_hour(three_minutes())    assert hours.to_dict() == {9: 200, 10: 200}

"""The watchlist, and a table of its quotes refreshed every minute. Usage:     uv run --env-file .env 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"]  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()

"""Download a symbol's daily bars since 2016, save them, and chart the close. Usage:     uv run --env-file .env history AAPL""" import argparseimport sys import matplotlib.pyplot as pltimport pandas as pdfrom matplotlib import ticker from market_data import alpacafrom market_data.files import CHARTS, DATA # Alpaca's short names for a bar's fields, and the names this project uses.COLUMNS = {"o": "open", "h": "high", "l": "low", "c": "close", "v": "volume"}  def get_history(symbol: str) -> pd.DataFrame:    """Every trading day's open, high, low, close and volume since 2016, adjusted    for splits, oldest first, dated."""    bars = alpaca.daily_bars(symbol)    prices = pd.DataFrame(bars, columns=list(COLUMNS)).rename(columns=COLUMNS)    prices.index = pd.DatetimeIndex(        [alpaca.day_of(bar["t"]) for bar in bars], name="date"    )    return prices  def chart(prices: pd.DataFrame, symbol: str, path: str) -> None:    """The daily close on a log scale, saved as a PNG."""    fig, ax = plt.subplots(figsize=(10, 5))    ax.plot(prices.index, prices["close"], linewidth=1)    ax.set_yscale("log")    ax.yaxis.set_major_locator(ticker.LogLocator(subs=[1, 2, 5]))    ax.yaxis.set_major_formatter(ticker.StrMethodFormatter("{x:,g}"))    ax.yaxis.set_minor_formatter(ticker.NullFormatter())    ax.grid(alpha=0.3)    ax.set_title(f"{symbol} daily close")    ax.set_ylabel("USD, log scale")    fig.savefig(path, dpi=120, bbox_inches="tight")  def main() -> None:    parser = argparse.ArgumentParser(        description="Download, save and chart a symbol's daily bars."    )    parser.add_argument(        "symbol", nargs="?", default="AAPL", help="ticker symbol (default: AAPL)"    )    symbol = parser.parse_args().symbol    prices = get_history(symbol)    if prices.empty:        sys.exit(f"No prices for {symbol}: check the ticker.")    print(prices.tail())    print(        f"\n{len(prices):,} days, {prices.index[0]:%d %b %Y} to {prices.index[-1]:%d %b %Y}"    )    best = prices["close"].idxmax()    print(f"highest close: {prices['close'].max():.2f} on {best:%d %b %Y}")    DATA.mkdir(exist_ok=True)    CHARTS.mkdir(exist_ok=True)    prices.to_csv(DATA / f"{symbol}.csv")    chart(prices, symbol, str(CHARTS / f"{symbol}.png"))    print(f"saved {DATA / symbol}.csv and {CHARTS / symbol}.png")  if __name__ == "__main__":    main()

"""Daily, total and yearly returns of a symbol, from the prices history saved. Usage:     uv run returns AAPL""" import argparse import pandas as pd from market_data.files import DATA  def load(symbol: str) -> pd.DataFrame:    """The prices history saved, dated."""    return pd.read_csv(DATA / f"{symbol}.csv", index_col="date", parse_dates=True)  def daily_returns(close: pd.Series) -> pd.Series:    """Each day's return: the close over the previous close, minus 1."""    return close.pct_change().dropna()  def total_return(close: pd.Series) -> float:    """The period's return: the last close over the first, minus 1."""    return float(close.iloc[-1] / close.iloc[0] - 1)  def yearly_growth(close: pd.Series) -> float:    """The compound annual growth rate: the yearly return that compounds to the total."""    years = (close.index[-1] - close.index[0]).days / 365.25    return float((close.iloc[-1] / close.iloc[0]) ** (1 / years) - 1)  def calendar_years(close: pd.Series) -> pd.Series:    """Each calendar year's return, from one year's last close to the next."""    return close.resample("YE").last().pct_change().dropna()  def main() -> None:    parser = argparse.ArgumentParser(        description="Print a symbol's returns from its saved prices."    )    parser.add_argument(        "symbol", nargs="?", default="AAPL", help="ticker symbol (default: AAPL)"    )    close = load(parser.parse_args().symbol)["close"]    daily = daily_returns(close)    print(f"best day:  {daily.max():+.2%} on {daily.idxmax():%d %b %Y}")    print(f"worst day: {daily.min():+.2%} on {daily.idxmin():%d %b %Y}")    print(f"from {close.iloc[0]:.2f} to {close.iloc[-1]:.2f}")    print(f"total return: {total_return(close):+,.1%}")    print(f"growth a year, compounded: {yearly_growth(close):+.1%}")    print("\nyear   return")    for year, r in calendar_years(close).items():        print(f"{year:%Y}  {r:+7.1%}")  if __name__ == "__main__":    main()

"""Find splits in raw prices, adjust for them, and count dividends in. Reads the raw sample in data/split-sample.csv and your prices in data/AAPL.csv. Usage:     uv run adjust""" import math import pandas as pd from market_data.files import DATAfrom market_data.returns import total_return # A 2-for-1 split halves the price, and the stock may rise a little that day too.# A fall of more than 40% in a day is almost always a split, not a crash.SPLIT_MOVE = -0.4  def find_splits(close: pd.Series) -> dict[pd.Timestamp, int]:    """Days the close fell by more than 40%, each with its split ratio."""    returns = close.pct_change()    days = returns[returns < SPLIT_MOVE].index    return {day: round(close.shift(1)[day] / close[day]) for day in days}  def adjust(prices: pd.DataFrame, splits: dict[pd.Timestamp, int]) -> pd.DataFrame:    """Divide each price and dividend before a split by its ratio; multiply volume."""    adjusted = prices.copy()    for day, ratio in splits.items():        before = adjusted.index < day        adjusted.loc[before, ["close", "dividend"]] /= ratio        adjusted.loc[before, "volume"] *= ratio    return adjusted  def total_returns(close: pd.Series, dividend: pd.Series) -> pd.Series:    """Each day's return with that day's dividend included."""    return ((close + dividend) / close.shift(1) - 1).dropna()  def main() -> None:    raw = pd.read_csv(DATA / "split-sample.csv", index_col="date", parse_dates=True)    splits = find_splits(raw["close"])    for day, ratio in splits.items():        print(f"split on {day:%d %b %Y}: {ratio} for 1")    prices = adjust(raw, splits)    close = prices["close"]    total = math.prod(1 + total_returns(close, prices["dividend"])) - 1    print(f"from {close.iloc[0]:.4f} to {close.iloc[-1]:.4f}")    print(f"price return:                {total_return(close):+.2%}")    print(f"total return, with dividend: {total:+.2%}")    mine = pd.read_csv(DATA / "AAPL.csv", index_col="date", parse_dates=True)    print(f"splits in {DATA / 'AAPL.csv'}: {len(find_splits(mine['close']))}")  if __name__ == "__main__":    main()

"""A trading day's 1-minute bars: the day's bar, its VWAP and volume by hour. Usage:     uv run --env-file .env intraday AAPL    uv run --env-file .env intraday AAPL --day 2026-10-05 With no day, it shows the latest day the symbol traded.""" import argparseimport sysfrom datetime import date import matplotlib.dates as mdatesimport matplotlib.pyplot as pltimport pandas as pdfrom matplotlib import ticker from market_data import alpacafrom market_data.files import CHARTS, DATA # Alpaca's short names for a bar's fields, and the names this project uses.COLUMNS = {    "o": "open",    "h": "high",    "l": "low",    "c": "close",    "v": "volume",    "vw": "vwap",}  def get_minutes(symbol: str, day: date | None = None) -> pd.DataFrame:    """A day's 1-minute bars, oldest first, labelled by their start in New York    time: the latest day the symbol traded, if no day is given. Empty if Alpaca has    no such symbol, or the market wasn't open that day."""    day = day or alpaca.latest_day(symbol)    bars = alpaca.minute_bars(symbol, day) if day else []    minutes = pd.DataFrame(bars, columns=list(COLUMNS)).rename(columns=COLUMNS)    times = pd.to_datetime([bar["t"] for bar in bars], utc=True)    minutes.index = times.tz_convert(alpaca.NEW_YORK).rename("time")    return minutes  def day_bar(minutes: pd.DataFrame) -> dict[str, float]:    """The day as one bar, built from its minutes."""    return {        "open": float(minutes["open"].iloc[0]),        "high": float(minutes["high"].max()),        "low": float(minutes["low"].min()),        "close": float(minutes["close"].iloc[-1]),        "volume": float(minutes["volume"].sum()),    }  def vwap(minutes: pd.DataFrame) -> float:    """The day's volume-weighted average price, from each minute's own."""    traded = (minutes["vwap"] * minutes["volume"]).sum()    return float(traded / minutes["volume"].sum())  def volume_by_hour(minutes: pd.DataFrame) -> pd.Series:    """Shares traded in each clock hour. Hour 9 is the first half hour, 9:30 to 10:00."""    return minutes["volume"].groupby(pd.DatetimeIndex(minutes.index).hour).sum()  def chart(minutes: pd.DataFrame, symbol: str, path: str) -> None:    """Price and VWAP above, volume below, saved as a PNG."""    fig, (top, bottom) = plt.subplots(        2, 1, figsize=(10, 6), sharex=True, height_ratios=[3, 1]    )    top.plot(minutes.index, minutes["close"], linewidth=1, label="price")    top.axhline(vwap(minutes), color="tab:orange", linestyle="--", label="VWAP")    top.set_title(f"{symbol} 1-minute bars, {minutes.index[0]:%d %b %Y}")    top.set_ylabel("USD")    top.legend()    bottom.bar(minutes.index, minutes["volume"], width=1 / (24 * 60), color="gray")    bottom.set_ylabel("shares")    bottom.yaxis.set_major_formatter(ticker.StrMethodFormatter("{x:,.0f}"))    bottom.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M", tz=alpaca.NEW_YORK))    fig.savefig(path, dpi=120, bbox_inches="tight")  def main() -> None:    parser = argparse.ArgumentParser(        description="Summarise and chart a trading day's 1-minute bars."    )    parser.add_argument(        "symbol", nargs="?", default="AAPL", help="ticker symbol (default: AAPL)"    )    parser.add_argument(        "--day", type=date.fromisoformat, help="YYYY-MM-DD (default: the latest)"    )    args = parser.parse_args()    symbol = args.symbol    minutes = get_minutes(symbol, args.day)    if minutes.empty:        sys.exit(f"No minutes for {symbol} that day: check the ticker and the day.")    first, last = minutes.index[0], minutes.index[-1]    print(        f"{len(minutes)} minutes on {first:%d %b %Y}, {first:%H:%M} to {last:%H:%M}\n"    )    bar = day_bar(minutes)    for field in ("open", "high", "low", "close"):        print(f"{field:<8}{bar[field]:.2f}")    print(f"{'volume':<8}{bar['volume']:,.0f}")    print(f"{'VWAP':<8}{vwap(minutes):.2f}")    hours = volume_by_hour(minutes)    print("\nhour      shares")    for hour, shares in hours.items():        print(f"{hour:>4}{shares:>12,}  {'#' * round(30 * shares / hours.max())}")    DATA.mkdir(exist_ok=True)    CHARTS.mkdir(exist_ok=True)    day = f"{symbol}-{first:%Y-%m-%d}"    minutes.to_csv(DATA / f"{day}.csv")    chart(minutes, symbol, str(CHARTS / f"{day}.png"))    print(f"\nsaved {DATA / day}.csv and {CHARTS / day}.png")  if __name__ == "__main__":    main()