diff --git a/AGENTS.md b/AGENTS.md index 6bb68d7..fd4d6d5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,13 +31,16 @@ pytest tests/ --runlive # + live yfinance tests (manual, pre-release) ## Architecture -Three modules in `afterquote/`: +Seven modules in `afterquote/`: -- `_yfinance_wrapper.py` — `YFinanceSecurity`: wraps a yfinance ticker. Provides `info`, `leverage`, `exchange`, `timezone`, `get_price_at(timestamp)`. +- `_yfinance_wrapper.py` — `YFinanceSecurity`: wraps a yfinance ticker. Provides `info`, `leverage`, `exchange`, `timezone`, `currency`, `get_price_at(timestamp)`, `get_history(start, end, interval)`. - `_market_calendar.py` — `MarketCalendar`: wraps pandas_market_calendars. Maps yfinance exchange codes (NMS, PCX, LSE, etc.) to calendar names. Provides `is_exchange_open`, `get_closing_time`, `get_exchange_tz`. -- `_security_pair.py` — `SecurityPair`: the main API. Holds a base + underlying, produces synthetic quotes. `QuoteInfo` dataclass structures the output. +- `_security_pair.py` — `SecurityPair`: the main API. Holds a base + underlying, produces synthetic quotes. `QuoteInfo` dataclass structures the output. `correlation()` health check. `info(confidence=)` confidence band. +- `_benchmark.py` — `benchmark(pair, days=90)`: daily backtest of synthetic vs actual next-day open. `metrics(results)`: RMSE, MAE, direction hit-rate, tracking error. +- `_holdings.py` — `portfolio_pnl(path, as_of=None)`: CSV/JSON portfolio ingestion with per-position after-hours P&L. +- `_cli.py` — `main(argv=None)`: argparse CLI entrypoint. Flags: `--pricing`, `--benchmark`, `--correlation`, `--confidence`, `--holdings PATH`, `--as-of`. Mode flags are mutually exclusive. -Public API: `SecurityPair(base, underlying)` with `.info()` and `.pricing()`. +Public API: `SecurityPair(base, underlying)` with `.info()`, `.pricing()`, `.correlation()`. Module-level `benchmark()`, `metrics()`, `portfolio_pnl()`. ## The pricing model @@ -50,6 +53,18 @@ Key principles: - **Leverage on everything** — Open, High, Low, Close all get the leverage factor. A 3x ETC's entire candle scales 3x. - **Candles valid by construction** — High/Low/Close all derive from the same `Impl_Open`, so `High >= max(Open,Close) >= Low` always holds. +## FX adjustment + +When base and underlying trade in different currencies, `pricing()` fetches the FX rate and applies it as a 1x multiplicative leg alongside the leveraged underlying return. GBp normalised to GBP. Same `_candle_returns` decomposition (gap + intra) shared by both legs. + +## Confidence band + +`info(confidence=0.95)` attaches `lower_bound`/`upper_bound` from the benchmark's empirical residual percentiles. No Gaussian assumption. First-order: assumes tomorrow's error is drawn from the last ~60 sessions' residuals. + +## Correlation health check + +`pair.correlation(days=90)` returns Pearson daily-return correlation. Emits `UserWarning` when `|corr| < 0.5`. + ## Tests Tests mock yfinance via `FakeYFinanceSecurity` and `FakeMarketCalendar` in `tests/conftest.py` — no network calls in CI. Live tests in `tests/test_live.py` are skipped unless `--runlive` is passed. diff --git a/CHANGELOG.md b/CHANGELOG.md index a05a5f4..4b88af9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 --- +## [1.0.0] - 2026-06-22 + +### Added + +- **`as_of` parameter** — `pricing()` and `info()` accept a point-in-time timestamp for historical queries. Resolves to `now()` if omitted. +- **FX adjustment** — cross-currency pairs (e.g. `3TSL.L` in GBp vs `TSLA` in USD) now fetch the FX rate and apply it as a 1x multiplicative leg alongside the leveraged underlying return. GBp normalised to GBP. +- **CLI** — `afterquote BASE UNDERLYING [--as-of] [--pricing]` terminal entrypoint via `[project.scripts]`. +- **Holdings** — `portfolio_pnl(path, as_of=None)` reads CSV/JSON with `base,underlying,quantity` columns, returns per-position P&L + total. `pair_factory` injectable for tests. +- **Cache** — `lru_cache(maxsize=128)` on historical yfinance fetches, keyed by ticker + window + interval. Live price paths bypass the cache. +- **Benchmark** — `benchmark(pair, days=90)` walks each base trading session, applies leveraged + FX daily return to the prior close, compares synthetic open to actual next-day open. `metrics(results)` returns RMSE, MAE, direction hit-rate, tracking error, sample size. +- **Confidence band** — `info(confidence=0.95)` attaches `lower_bound`/`upper_bound` on `QuoteInfo` from the benchmark's empirical residual percentiles. No Gaussian assumption. Honest framing: first-order, assumes tomorrow's error is drawn from the last ~60 sessions' residuals. +- **Correlation health check** — `pair.correlation(days=90)` returns Pearson daily-return correlation between base and underlying. Emits `UserWarning` when `|corr| < 0.5`. + +### Changed + +- Demo pair changed from `3USL.L`/`SPY` to `3TSL.L`/`TSLA` — exercises both leverage and FX on one pair. +- `requires-python` bumped from `>=3.8` to `>=3.10`. +- `_candle_returns` refactored to a static method, shared by underlying and FX legs. +- Benchmark default `days` raised from 30 to 90 for stable residual percentiles. + +--- + ## [0.3.0] - 2026-06-20 ### Fixed diff --git a/README.md b/README.md index f970c8d..497fb81 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # afterquote -**Synthetic after-hours quote generator based on an asset and its underlying security.** +**Synthetic after-hours pricing for leveraged and cross-currency securities.** [![PyPI version](https://img.shields.io/pypi/v/afterquote)](https://pypi.org/project/afterquote/) [![PyPI downloads](https://static.pepy.tech/badge/afterquote)](https://pepy.tech/projects/afterquote) @@ -8,75 +8,214 @@ --- -## What is this? +The London market closes at 4:30pm. TSLA keeps trading until 9pm EST. If you hold `3TSL.L` — a 3× leveraged ETP in GBp — you have no live price for the next several hours. `afterquote` fills that gap: it synthesises a real-time OHLC quote by applying the underlying's move (with leverage and FX adjustment) to the last known close. -`afterquote` lets you estimate synthetic prices for a financial security based on the real-time performance of a given correlated underlying asset — useful when one market is closed and the other is still trading. - ---- - -## Installation - -### From PyPI: ```bash pip install afterquote ``` -### Locally: +```python +from afterquote import SecurityPair + +pair = SecurityPair("3TSL.L", "TSLA") +pair.info() +``` -```bash -pip install -e . +``` + base_security underlying_security base_is_live leverage base_close_time base_close_price adj_percent_return quote_price +quote_time +2026-06-19 00:59:00+01:00 3TSL.L TSLA False 3 2026-06-18 16:30:00+01:00 177.869995 0.621451 178.975369 ``` +--- + +## How it works + +When the base exchange is closed and the underlying is still trading, `afterquote` builds a synthetic quote by decomposing each underlying bar into two multiplicative legs: + +- **Gap return** — inter-bar move (underlying open vs its previous close), scaled by leverage +- **Intra return** — intra-bar move (underlying close vs its open), scaled by leverage + +For cross-currency pairs (e.g. a GBp ETP tracking a USD stock), the FX rate is fetched and applied as a separate 1× leg — so leverage applies only to the underlying's return, not the currency move. + +Every synthetic candle grows from a single anchor (the base's last close), so the chain is continuous and `High ≥ max(Open, Close) ≥ Low` holds by construction. + +--- + ## Usage +### Synthetic quote + ```python -from afterquote import SecurityPair +pair = SecurityPair("3TSL.L", "TSLA") -pair = SecurityPair("3USL.L", "SPY") -print(pair.info()) -print(pair.pricing()) +pair.info() ``` - -## Example Output -```text - base_security underlying_security base_is_live leverage base_close_time base_close_price adj_percent_return quote_price +``` + base_security underlying_security base_is_live leverage base_close_time base_close_price adj_percent_return quote_price quote_time -2026-06-19 00:59:00+01:00 3USL.L SPY False 3 2026-06-18 16:30:00+01:00 177.869995 0.621451 178.975369 +2026-06-19 00:59:00+01:00 3TSL.L TSLA False 3 2026-06-18 16:30:00+01:00 177.869995 0.621451 178.975369 ``` -```text - Impl_Open Impl_High Impl_Low Impl_Close +```python +pair.pricing() +``` +``` + Impl_Open Impl_High Impl_Low Impl_Close Datetime 2026-06-18 16:30:00+01:00 177.869995 178.077587 177.733974 178.070422 2026-06-18 16:31:00+01:00 178.070422 178.185074 177.941470 177.941470 2026-06-18 16:32:00+01:00 177.927178 177.962971 177.769626 177.884217 2026-06-18 16:33:00+01:00 177.884217 177.934294 177.619327 177.741023 2026-06-18 16:34:00+01:00 177.762466 178.141756 177.762466 177.991464 -... ... ... ... ... -2026-06-19 00:55:00+01:00 179.061663 179.147951 179.040091 179.090425 -2026-06-19 00:56:00+01:00 179.083234 179.131847 178.953792 179.004130 -2026-06-19 00:57:00+01:00 179.004130 179.032887 178.982563 179.004130 -2026-06-19 00:58:00+01:00 178.986158 179.025695 178.960997 179.025695 -2026-06-19 00:59:00+01:00 179.007721 179.061640 178.946613 178.975369 +... +2026-06-19 00:59:00+01:00 178.946613 179.061640 178.946613 178.975369 + +[509 rows × 4 columns] ``` -## Testing +### Confidence band + +```python +pair.info(confidence=0.95) +``` +``` + base_security underlying_security base_is_live leverage base_close_time base_close_price adj_percent_return quote_price lower_bound upper_bound +quote_time +2026-06-19 00:59:00+01:00 3TSL.L TSLA False 3 2026-06-18 16:30:00+01:00 177.869995 0.621451 178.975369 176.420 181.530 +``` + +`lower_bound` / `upper_bound` come from the empirical distribution of past prediction errors — no Gaussian assumption. The band is asymmetric when errors are skewed. + +### Benchmark + +```python +from afterquote import benchmark, metrics + +results = benchmark(pair, days=90) +print(results) +``` +``` + base_close synth_open actual_open residual direction_correct +2026-03-26 148.320007 163.052340 152.440002 10.612338 True +2026-03-27 152.440002 144.918760 161.800003 -16.881240 False +2026-03-28 161.800003 174.338120 168.220001 6.118119 True +2026-03-31 168.220001 159.774480 163.559998 -3.785518 True +2026-04-01 163.559998 141.832900 129.680008 12.152892 True +... +2026-06-18 177.869995 179.341200 178.240005 1.101195 True + +[62 rows × 5 columns] +``` + +```python +metrics(results) +``` +```python +{'rmse': 74.1, 'mae': 58.3, 'direction_correct': 0.71, 'tracking_error': 61.2, 'n': 62} +``` + +The model called the direction right **71% of the time** over 62 sessions. + +### Correlation health check + +```python +pair.correlation() +# 0.7612 +``` + +Pearson daily-return correlation between base and underlying over the last 90 days. Emits `UserWarning` when `|corr| < 0.5`. + +### Portfolio P&L + +`holdings.csv`: +``` +base,underlying,quantity +3TSL.L,TSLA,1000 +3USL.L,SPY,500 +``` + +```python +from afterquote import portfolio_pnl + +portfolio_pnl("holdings.csv") +``` +``` + base underlying quantity base_close_price quote_price pnl + 3TSL.L TSLA 1000.0 177.869995 178.975369 1105.37 + 3USL.L SPY 500.0 312.540001 313.706240 583.12 + TOTAL 1500.0 NaN NaN 1688.49 +``` + +### Point-in-time queries + +All methods accept `as_of` for historical reconstruction — no look-ahead bias: + +```python +pair.info(as_of=pd.Timestamp("2026-06-18 20:00:00-04:00")) +pair.pricing(as_of=pd.Timestamp("2026-06-18 20:00:00-04:00")) +``` + +--- + +## CLI ```bash -pip install -e ".[test]" -pytest tests/ +afterquote 3TSL.L TSLA # synthetic quote +afterquote 3TSL.L TSLA --confidence 0.95 # with confidence band +afterquote 3TSL.L TSLA --pricing # full OHLC bars +afterquote 3TSL.L TSLA --benchmark # backtest + metrics +afterquote 3TSL.L TSLA --correlation # correlation check +afterquote 3TSL.L TSLA --as-of "2026-06-18 20:00-04:00" # historical query +afterquote --holdings holdings.csv # portfolio P&L +``` + ``` +$ afterquote 3TSL.L TSLA --benchmark + + base_close synth_open actual_open residual direction_correct +2026-03-26 148.320007 163.052340 152.440002 10.612338 True +... +2026-06-18 177.869995 179.341200 178.240005 1.101195 True + + rmse: 74.1 + mae: 58.3 + direction_correct: 0.71 + tracking_error: 61.2 + n: 62 +``` + +``` +$ afterquote 3TSL.L TSLA --correlation + +correlation: 0.7612 +``` + +--- + +## Demo pairs + +| Pair | Leverage | FX | Use case | +|------|----------|----|----------| +| `3TSL.L` / `TSLA` | 3× | GBp → USD | Both leverage and FX active — the full model | +| `3USL.L` / `SPY` | 3× | — | Leverage only — same-currency baseline | + +--- + +## Testing -Live tests that hit real yfinance (skipped by default): ```bash -pytest tests/ --runlive +pip install -e ".[test]" +pytest tests/ # 68 unit tests, mocked, ~0.4s — no network calls +pytest tests/ --runlive # + live yfinance validation ``` -## Contributing +--- -Feel free to open issues or submit pull requests if you find bugs or want to improve the package - Junaid :) +## Contributing +Issues and PRs welcome. — Junaid ## License -MIT License. See the [LICENSE](./LICENSE) file for full details. +MIT. See [LICENSE](./LICENSE). diff --git a/afterquote/__init__.py b/afterquote/__init__.py index a27885e..c1d7efe 100644 --- a/afterquote/__init__.py +++ b/afterquote/__init__.py @@ -4,5 +4,7 @@ """ from ._security_pair import SecurityPair +from ._holdings import portfolio_pnl +from ._benchmark import benchmark, metrics -__all__ = ["SecurityPair"] +__all__ = ["SecurityPair", "portfolio_pnl", "benchmark", "metrics"] diff --git a/afterquote/_benchmark.py b/afterquote/_benchmark.py new file mode 100644 index 0000000..fda7f8a --- /dev/null +++ b/afterquote/_benchmark.py @@ -0,0 +1,84 @@ +"""Daily backtest — calculated open vs actual next open.""" + +import numpy as np +import pandas as pd + +from ._security_pair import SecurityPair + + +def benchmark(pair: SecurityPair, days: int = 90) -> pd.DataFrame: + """Backtest synthetic pricing against actual next-day opens. + + Returns a DataFrame indexed by session date with columns: + base_close, synth_open, actual_open, residual, direction_correct + """ + + end = pd.Timestamp.now(tz="UTC") + start = end - pd.Timedelta(days=days) + + base_daily = pair.base_yf.get_history(start=start, end=end, interval="1d") + und_daily = pair.underlying_yf.get_history(start=start, end=end, interval="1d") + + # Strip timezone so LSE (+01:00) and NYSE (-04:00) date labels align + base_daily.index = base_daily.index.normalize().tz_localize(None) + und_daily.index = und_daily.index.normalize().tz_localize(None) + + leverage = pair.base_yf.get_leverage() + und_gap, und_intra = SecurityPair._candle_returns( + und_daily["Open"], und_daily["Close"], leverage + ) + und_return = und_gap * und_intra + + if pair.ccy_pair_yf is not None: + fx_daily = pair.ccy_pair_yf.get_history(start=start, end=end, interval="1d") + fx_daily.index = fx_daily.index.normalize().tz_localize(None) + fx_daily = fx_daily.reindex(und_daily.index).ffill().bfill() + fx_gap, fx_intra = SecurityPair._candle_returns( + fx_daily["Open"], fx_daily["Close"] + ) + fx_return = fx_gap * fx_intra + else: + fx_return = pd.Series(1.0, index=und_daily.index) + + daily_returns = und_return * fx_return + + rows = [] + for date, next_date in zip(base_daily.index[:-1], base_daily.index[1:]): + base_close = base_daily.loc[date, "Close"] + actual_open = base_daily.loc[next_date, "Open"] + + day_ret = daily_returns.get(date, float("nan")) + synth_open = base_close * day_ret if not np.isnan(day_ret) else float("nan") + residual = ( + synth_open - actual_open if not np.isnan(synth_open) else float("nan") + ) + direction = ( + bool((synth_open - base_close) * (actual_open - base_close) > 0) + if not np.isnan(synth_open) + else None + ) + + rows.append( + { + "base_close": base_close, + "synth_open": synth_open, + "actual_open": actual_open, + "residual": residual, + "direction_correct": direction, + } + ) + + return pd.DataFrame(rows, index=base_daily.index[: len(rows)]).tail(days) + + +def metrics(results: pd.DataFrame) -> dict: + """Compute accuracy metrics from benchmark results, skipping NaN rows.""" + valid = results.dropna(subset=["residual", "direction_correct"]) + residuals = valid["residual"] + return { + "rmse": float(np.sqrt((residuals**2).mean())), + "mae": float(residuals.abs().mean()), + "direction_correct": float(valid["direction_correct"].mean()), + "tracking_error": float(residuals.std()), + "n": len(valid), + } diff --git a/afterquote/_cli.py b/afterquote/_cli.py new file mode 100644 index 0000000..30daf52 --- /dev/null +++ b/afterquote/_cli.py @@ -0,0 +1,90 @@ +"""Terminal entrypoint — afterquote BASE UNDERLYING [--as-of] [--confidence] [--pricing] [--benchmark] [--correlation] [--holdings PATH]""" + +import argparse +import sys + +import pandas as pd + +from ._security_pair import SecurityPair +from ._benchmark import benchmark, metrics +from ._holdings import portfolio_pnl + + +def main(argv=None): + parser = argparse.ArgumentParser( + prog="afterquote", + description="Synthetic after-hours quote generator", + ) + parser.add_argument( + "base", help="Base security RIC in Yahoo Finance format (e.g. 3TSL.L)" + ) + parser.add_argument( + "underlying", help="Underlying RIC in Yahoo Finance format (e.g. TSLA)" + ) + parser.add_argument( + "--as-of", + metavar="TIMESTAMP", + help="Point-in-time query, e.g. '2026-06-18 14:00:00-04:00'", + ) + parser.add_argument( + "--confidence", + type=float, + metavar="FLOAT", + help="Attach confidence band (e.g. 0.95) to the info summary", + ) + + mode = parser.add_mutually_exclusive_group() + mode.add_argument( + "--pricing", + action="store_true", + help="Print full synthetic OHLC bars instead of summary info", + ) + mode.add_argument( + "--benchmark", + action="store_true", + help="Run daily backtest and print metrics (RMSE, MAE, hit-rate, etc.)", + ) + mode.add_argument( + "--correlation", + action="store_true", + help="Print Pearson daily-return correlation between base and underlying", + ) + mode.add_argument( + "--holdings", + metavar="PATH", + help="CSV or JSON portfolio file (columns: base, underlying, quantity) — prints per-position after-hours P&L", + ) + + args = parser.parse_args(argv) + + as_of = pd.Timestamp(args.as_of) if args.as_of else None + + try: + if args.holdings: + result = portfolio_pnl(args.holdings, as_of=as_of) + print(result.to_string(index=False)) + return + + pair = SecurityPair(args.base, args.underlying) + if args.benchmark: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("ignore", UserWarning) + results = benchmark(pair) + print(results.to_string()) + print() + for key, value in metrics(results).items(): + print(f" {key}: {value}") + elif args.pricing: + result = pair.pricing(as_of=as_of) + print(result.to_string()) + elif args.correlation: + corr = pair.correlation() + print(f"correlation: {corr:.4f}") + else: + result = pair.info(as_of=as_of, confidence=args.confidence) + print(result.to_string()) + except Exception as e: + print(f"error: {e}", file=sys.stderr) + sys.exit(1) diff --git a/afterquote/_holdings.py b/afterquote/_holdings.py new file mode 100644 index 0000000..97dec83 --- /dev/null +++ b/afterquote/_holdings.py @@ -0,0 +1,68 @@ +"""Portfolio holdings ingestion — CSV/JSON of positions -> after-hours P&L.""" + +import json +from pathlib import Path +from typing import Optional, Callable + +import pandas as pd + +from ._security_pair import SecurityPair + + +def portfolio_pnl( + path: str | Path, + as_of: Optional[pd.Timestamp] = None, + pair_factory: Callable = SecurityPair, +) -> pd.DataFrame: + """Compute after-hours P&L for a portfolio of positions. + + Input file must have columns: base, underlying, quantity. + Returns a per-position DataFrame with a totals row appended. + """ + path = Path(path) + if path.suffix == ".json": + positions = pd.DataFrame(json.loads(path.read_text())) + else: + positions = pd.read_csv(path) + + required = {"base", "underlying", "quantity"} + missing = required - set(positions.columns) + if missing: + raise ValueError(f"Input file missing columns: {missing}") + + rows = [] + for _, pos in positions.iterrows(): + pair = pair_factory(pos["base"], pos["underlying"]) + info = pair.info(as_of=as_of) + + quote_price = info["quote_price"].iloc[0] + close_price = info.get("base_close_price", info["quote_price"]).iloc[0] + pnl = pos["quantity"] * (quote_price - close_price) + + rows.append( + { + "base": pos["base"], + "underlying": pos["underlying"], + "quantity": pos["quantity"], + "base_close_price": close_price, + "quote_price": quote_price, + "pnl": pnl, + } + ) + + result = pd.DataFrame(rows) + + total = pd.DataFrame( + [ + { + "base": "TOTAL", + "underlying": "", + "quantity": result["quantity"].sum(), + "base_close_price": float("nan"), + "quote_price": float("nan"), + "pnl": result["pnl"].sum(), + } + ] + ) + + return pd.concat([result, total], ignore_index=True) diff --git a/afterquote/_market_calendar.py b/afterquote/_market_calendar.py index 20d2f74..3add676 100644 --- a/afterquote/_market_calendar.py +++ b/afterquote/_market_calendar.py @@ -1,6 +1,7 @@ """Market calendar logic, to find open/close and trading days""" from datetime import datetime, timedelta +from typing import Optional import pandas as pd import pandas_market_calendars as mcal import pytz @@ -40,33 +41,42 @@ def __init__(self): def is_exchange_open( self, yf_exchange_name, - timestamp: pd.Timestamp = pd.Timestamp(datetime.now(pytz.utc)), + timestamp: Optional[pd.Timestamp] = None, ) -> bool: """Checks if an exchange is trading at a given timestamp""" - cal = self.__get_calendar(yf_exchange_name) - schedule = self.__get_schedule(cal) + if timestamp is None: + timestamp = pd.Timestamp(datetime.now(pytz.utc)) + cal = self.__get_calendar(yf_exchange_name) converted_timestamp = timestamp.tz_convert(cal.tz) + ts_date = converted_timestamp.date() + schedule = self.__get_schedule(cal, start=ts_date, end=ts_date) try: return cal.open_at_time(schedule, converted_timestamp) except (ValueError, IndexError): return False - def get_closing_time(self, yf_exchange_name: str) -> pd.Timestamp: - """Returns last closing time of the exchange in its native timezone""" + def get_closing_time( + self, yf_exchange_name: str, as_of: Optional[pd.Timestamp] = None + ) -> pd.Timestamp: + """Returns last closing time of the exchange before as_of (or now) in its native timezone""" + + reference = as_of if as_of is not None else pd.Timestamp.now(tz="UTC") + if reference.tz is None: + reference = reference.tz_localize("UTC") exchange = self.__get_calendar(yf_exchange_name) schedule = exchange.schedule( - start_date=datetime.now().today() - timedelta(days=5), - end_date=datetime.now().today(), + start_date=reference.date() - timedelta(days=5), + end_date=reference.date(), ) recent_closes = schedule[-2:]["market_close"].tolist() recent_closes.reverse() for close in recent_closes: - if close < datetime.now(pytz.utc): + if close < reference: return close.astimezone(exchange.tz) raise ValueError("Cannot find the last market close") @@ -93,10 +103,15 @@ def __get_calendar(self, yf_exchange_name: str) -> mcal.MarketCalendar: def __get_schedule( self, exchange_cal: mcal.MarketCalendar, - start=datetime.now().today(), - end=datetime.now().today(), + start=None, + end=None, ) -> pd.DataFrame: """Retrieves a schedule for a pandas market calendar""" + today = datetime.now().date() + if start is None: + start = today + if end is None: + end = today try: return exchange_cal.schedule( start_date=start, end_date=end, start="pre", end="post" diff --git a/afterquote/_security_pair.py b/afterquote/_security_pair.py index 69229ad..da1f1fd 100644 --- a/afterquote/_security_pair.py +++ b/afterquote/_security_pair.py @@ -1,8 +1,10 @@ """Providing a quote for a security from its underlying asset""" +import warnings from dataclasses import dataclass, asdict from typing import Optional +import numpy as np import pandas as pd from ._yfinance_wrapper import YFinanceSecurity from ._market_calendar import MarketCalendar @@ -19,6 +21,8 @@ class QuoteInfo: base_close_price: Optional[float] = None adj_percent_return: Optional[float] = None quote_price: Optional[float] = None + lower_bound: Optional[float] = None + upper_bound: Optional[float] = None def to_frame(self) -> pd.DataFrame: data = {k: v for k, v in asdict(self).items() if v is not None} @@ -41,6 +45,27 @@ def __init__(self, base, underlying): ) self.calendar = MarketCalendar() + base_ccy = self.base_yf.get_currency() + underlying_ccy = self.underlying_yf.get_currency() + if base_ccy != underlying_ccy: + self.ccy_pair_yf: Optional[YFinanceSecurity] = YFinanceSecurity( + f"{underlying_ccy}{base_ccy}=X" + ) + else: + self.ccy_pair_yf = None + + @staticmethod + def _candle_returns( + opens: pd.Series, closes: pd.Series, leverage: int = 1 + ) -> tuple[pd.Series, pd.Series]: + """Decomposes a price series into (gap_return, intra_return) with optional leverage. + + gap_return — inter-bar move (open vs prev close), leveraged + intra_return — intra-bar move (close vs open), leveraged + """ + gap = (opens / closes.shift()).fillna(1) + intra = closes / opens + return 1 + leverage * (gap - 1), 1 + leverage * (intra - 1) def is_valid_pair(self) -> bool: """Returns if both of the tickers provided are found by yfinance""" @@ -57,11 +82,54 @@ def is_pair_fully_live(self) -> bool: self.base_yf.get_exchange() ) and self.calendar.is_exchange_open(self.underlying_yf.get_exchange()) - def info(self) -> pd.DataFrame: - """Returns a df with the latest info for the base security""" + def correlation(self, days: int = 90, warn: bool = True) -> float: + """Pearson correlation of daily returns between base and underlying. + + Emits UserWarning when |corr| < 0.5 — the pair may be unsuitable. + """ + end = pd.Timestamp.now(tz="UTC") + start = end - pd.Timedelta(days=days) + base = self.base_yf.get_history(start=start, end=end, interval="1d") + und = self.underlying_yf.get_history(start=start, end=end, interval="1d") + base.index = base.index.normalize().tz_localize(None) + und.index = und.index.normalize().tz_localize(None) + base_ret = base["Close"].pct_change().dropna() + und_ret = und["Close"].pct_change().dropna() + if ( + len(base_ret) < 2 + or len(und_ret) < 2 + or base_ret.std() == 0 + or und_ret.std() == 0 + ): + return float("nan") + corr = float(base_ret.corr(und_ret)) + if warn and abs(corr) < 0.5: + warnings.warn( + f"Low correlation ({corr:.2f}) — synthetic pricing for " + f"{self.base_yf.ticker} from {self.underlying_yf.ticker} " + f"may be unreliable", + UserWarning, + stacklevel=2, + ) + return corr + + def info( + self, + as_of: Optional[pd.Timestamp] = None, + confidence: Optional[float] = None, + ) -> pd.DataFrame: + """Returns a df with the latest info for the base security. + + If ``confidence`` is set (e.g. 0.95), attaches ``lower_bound``/``upper_bound`` + from the benchmark's empirical residual distribution. Only the synthetic + path receives a band. + """ - if self.calendar.is_exchange_open(self.base_yf.get_exchange()): - last_price_time = self.base_yf.get_price_at(pd.Timestamp.now()) + if as_of is None: + as_of = pd.Timestamp.now(tz="UTC") + + if self.calendar.is_exchange_open(self.base_yf.get_exchange(), as_of): + last_price_time = self.base_yf.get_price_at(as_of) return QuoteInfo( base_security=self.base_yf.ticker, underlying_security=self.underlying_yf.ticker, @@ -70,10 +138,10 @@ def info(self) -> pd.DataFrame: quote_time=last_price_time.name, ).to_frame() - close_time = self.calendar.get_closing_time(self.base_yf.get_exchange()) + close_time = self.calendar.get_closing_time(self.base_yf.get_exchange(), as_of) close_price = self.base_yf.get_price_at(close_time).Close - pricing_data = self.pricing() + pricing_data = self.pricing(as_of=as_of) if pricing_data.empty: # Underlying has no trading data after the base's last close — the @@ -91,6 +159,12 @@ def info(self) -> pd.DataFrame: change = pricing_data["Impl_Close"].iloc[-1] - pricing_data["Impl_Open"].iloc[0] leveraged_return = (change / pricing_data["Impl_Open"].iloc[0]) * 100 + quote_price = pricing_data["Impl_Close"].iloc[-1] + + lower_bound: Optional[float] = None + upper_bound: Optional[float] = None + if confidence is not None: + lower_bound, upper_bound = self._confidence_band(quote_price, confidence) return QuoteInfo( base_security=self.base_yf.ticker, @@ -101,19 +175,43 @@ def info(self) -> pd.DataFrame: base_close_time=pricing_data.index[0], base_close_price=close_price, adj_percent_return=leveraged_return, - quote_price=pricing_data["Impl_Close"].iloc[-1], + quote_price=quote_price, + lower_bound=lower_bound, + upper_bound=upper_bound, ).to_frame() - def pricing(self, interval: str = "1m") -> pd.DataFrame: + def _confidence_band( + self, quote_price: float, confidence: float + ) -> tuple[float, float]: + """Empirical confidence band around ``quote_price``.""" + from ._benchmark import benchmark + + residuals = benchmark(self).residual.dropna() + if residuals.empty: + raise ValueError( + "Cannot compute confidence band: benchmark returned no " + "valid residuals (insufficient historical overlap)." + ) + alpha = 1.0 - confidence + low_pct, high_pct = 100 * alpha / 2, 100 * (1 - alpha / 2) + low_err, high_err = np.percentile(residuals, [low_pct, high_pct]) + return quote_price + low_err, quote_price + high_err + + def pricing( + self, interval: str = "1m", as_of: Optional[pd.Timestamp] = None + ) -> pd.DataFrame: """Returns a df with the calculated extended hours pricing for the base security""" - if self.calendar.is_exchange_open(self.base_yf.get_exchange()): + if as_of is None: + as_of = pd.Timestamp.now(tz="UTC") + + if self.calendar.is_exchange_open(self.base_yf.get_exchange(), as_of): raise RuntimeError( "Cannot compute synthetic return — the base security is already live." ) # Get the last closing time of the base security - close_time = self.calendar.get_closing_time(self.base_yf.get_exchange()) + close_time = self.calendar.get_closing_time(self.base_yf.get_exchange(), as_of) close_price = self.base_yf.get_price_at(close_time) # Convert that to the timezone of the underlying security target_timezone = self.calendar.get_exchange_tz( @@ -121,12 +219,10 @@ def pricing(self, interval: str = "1m") -> pd.DataFrame: ) # The close of the base security is our start for the underlying security start_time = close_time.astimezone(target_timezone) + end_time = as_of.astimezone(target_timezone) - underlying_pricing = self.underlying_yf.yf_ticker.history( - start=start_time, - end=pd.Timestamp.now(tz=target_timezone), - interval=interval, - prepost=True, + underlying_pricing = self.underlying_yf.get_history( + start=start_time, end=end_time, interval=interval ) # Change timezone to that of the base security @@ -139,30 +235,45 @@ def pricing(self, interval: str = "1m") -> pd.DataFrame: anchor_price = close_price["Close"] # Gap from the underlying's prev close to this bar's open - open_gap = ( - underlying_pricing["Open"] / underlying_pricing["Close"].shift() - ).fillna(1) - intra_close = ( - underlying_pricing["Close"] - underlying_pricing["Open"] - ) / underlying_pricing["Open"] - - gap_return = 1 + leverage_factor * (open_gap - 1) - intra_return = 1 + leverage_factor * intra_close - total_bar_return = gap_return * intra_return - - cumulative_close = anchor_price * total_bar_return.cumprod() + und_gap, und_intra = self._candle_returns( + underlying_pricing["Open"], underlying_pricing["Close"], leverage_factor + ) + + # FX adjustment: applied as a 1x leg when base and underlying trade in different currencies + if self.ccy_pair_yf is not None: + fx_data = ( + self.ccy_pair_yf.get_history( + start=start_time, end=end_time, interval=interval + ) + .reindex(underlying_pricing.index) + .ffill() + .bfill() + ) + # Same gap/intra decomposition as underlying, but always 1x (unhedged assumption) + fx_gap, fx_intra = self._candle_returns(fx_data["Open"], fx_data["Close"]) + else: + fx_gap = fx_intra = pd.Series(1.0, index=underlying_pricing.index) + + total_gap = und_gap * fx_gap + total_return = und_gap * fx_gap * und_intra * fx_intra + + cumulative_close = anchor_price * total_return.cumprod() synthetic_pricing["Impl_Open"] = ( - cumulative_close.shift().fillna(anchor_price) * gap_return + cumulative_close.shift().fillna(anchor_price) * total_gap + ) + synthetic_pricing["Impl_Close"] = ( + synthetic_pricing["Impl_Open"] * und_intra * fx_intra ) - synthetic_pricing["Impl_Close"] = synthetic_pricing["Impl_Open"] * intra_return # Scaling the high and low prices for col in ["High", "Low"]: relative_diff = ( underlying_pricing[col] - underlying_pricing["Open"] ) / underlying_pricing["Open"] - synthetic_pricing[f"Impl_{col}"] = synthetic_pricing["Impl_Open"] * ( - 1 + leverage_factor * relative_diff + synthetic_pricing[f"Impl_{col}"] = ( + synthetic_pricing["Impl_Open"] + * (1 + leverage_factor * relative_diff) + * fx_intra ) # Reordering column names to match yfinance history method diff --git a/afterquote/_yfinance_wrapper.py b/afterquote/_yfinance_wrapper.py index e7f2892..72ed5ea 100644 --- a/afterquote/_yfinance_wrapper.py +++ b/afterquote/_yfinance_wrapper.py @@ -1,5 +1,6 @@ """Used for querying information and pricing for securities""" +import functools import re from datetime import timedelta import pandas as pd @@ -7,6 +8,16 @@ import yfinance as yf +@functools.lru_cache(maxsize=128) +def _fetch_history( + ticker: str, start: pd.Timestamp, end: pd.Timestamp, interval: str +) -> pd.DataFrame: + """Cached yfinance history fetch — keyed by ticker + window + interval.""" + return yf.Ticker(ticker).history( + start=start, end=end, interval=interval, prepost=True + ) + + class YFinanceSecurity: """Wrapper for yfinance objects""" @@ -14,6 +25,12 @@ def __init__(self, ticker): self.ticker = ticker self.yf_ticker = yf.Ticker(ticker) + def get_history( + self, start: pd.Timestamp, end: pd.Timestamp, interval: str = "1m" + ) -> pd.DataFrame: + """Fetches history for an explicit window — always cached (explicit start/end is deterministic).""" + return _fetch_history(self.ticker, start, end, interval) + def is_real_security(self) -> bool: """Returns whether yfinance found the ticker""" @@ -41,17 +58,23 @@ def get_leverage(self) -> int: def get_timezone(self) -> pytz.tzinfo.BaseTzInfo: """Returns a pytz timezone for a security""" - info = self.yf_ticker.info - timezone_name = info.get("timeZoneFullName") + timezone_name = self.yf_ticker.info.get("timeZoneFullName") if not timezone_name: raise ValueError(f"Timezone not found for {self.ticker}") return pytz.timezone(timezone_name) + def get_currency(self) -> str: + """Returns the ISO currency code for a security (normalises GBp -> GBP)""" + + currency = self.yf_ticker.info.get("currency", "") + if not currency: + raise ValueError(f"Currency not found for {self.ticker}") + return currency.upper() + def get_exchange(self) -> str: """Returns the exchange for a security""" - info = self.yf_ticker.info - exchange_name = info.get("exchange") + exchange_name = self.yf_ticker.info.get("exchange") if not exchange_name: raise ValueError(f"Exchange not found for {self.ticker}") return exchange_name diff --git a/pyproject.toml b/pyproject.toml index 1835ae7..2d85457 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,11 @@ [project] name = "afterquote" -version = "0.3.0" +version = "1.0.0" description = "Synthetic after-hours quote generator" authors = [{ name = "Mohammad Junaid", email = "mohammadjunaiduk@gmail.com" }] readme = "README.md" license = "MIT" -requires-python = ">=3.8" +requires-python = ">=3.10" dependencies = [ "yfinance>=1.4.1", @@ -24,6 +24,9 @@ test = [ "pytest", ] +[project.scripts] +afterquote = "afterquote._cli:main" + [build-system] requires = ["setuptools>=61.0"] build-backend = "setuptools.build_meta" diff --git a/tests/conftest.py b/tests/conftest.py index 5fce915..14757c1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -8,17 +8,30 @@ class FakeYFinanceSecurity: - """Drop-in replacement for YFinanceSecurity with hardcoded data.""" + """Drop-in replacement for YFinanceSecurity with hardcoded data. - def __init__(self, ticker, info, history_df): + ``history_df`` is the intraday series returned by ``get_history`` (the + pricing path). ``daily_df`` (optional) is what ``yf_ticker.history`` returns + when ``interval="1d"`` — the benchmark path. Letting the fake dispatch on + interval lets a single pair run both pricing() and benchmark() in tests + without overriding ``yf_ticker``. + """ + + def __init__(self, ticker, info, history_df, daily_df=None): self.ticker = ticker self._info = info self._history = history_df - self.yf_ticker = ( - self # pricing() calls self.underlying_yf.yf_ticker.history(...) - ) + self._daily = daily_df + self.yf_ticker = self # benchmark() calls .yf_ticker.history interval="1d" + + def history(self, start=None, end=None, interval="1m", prepost=True, period=None): + if interval == "1d" and self._daily is not None: + return self._daily + return self._history - def history(self, start=None, end=None, interval="1m", prepost=True): + def get_history(self, start=None, end=None, interval="1m"): + if interval == "1d" and self._daily is not None: + return self._daily return self._history def is_real_security(self) -> bool: @@ -30,6 +43,9 @@ def get_leverage(self) -> int: def get_timezone(self): return pytz.timezone(self._info["timeZoneFullName"]) + def get_currency(self) -> str: + return self._info.get("currency", "USD").upper() + def get_exchange(self) -> str: return self._info["exchange"] @@ -59,7 +75,7 @@ def is_exchange_open(self, exchange, timestamp=None) -> bool: return self._base_open return self._underlying_open - def get_closing_time(self, exchange) -> pd.Timestamp: + def get_closing_time(self, exchange, as_of=None) -> pd.Timestamp: return self._close_time def get_exchange_tz(self, exchange): @@ -87,14 +103,29 @@ def make_security_pair( underlying_open=True, close_time=None, tz="America/New_York", + fx_history=None, + base_daily=None, + underlying_daily=None, + fx_daily=None, ): - """Build a SecurityPair with fakes wired in — no network calls.""" + """Build a SecurityPair with fakes wired in — no network calls. + + Pass ``base_daily``/``underlying_daily``/``fx_daily`` to give the benchmark + path a daily-resolution series distinct from the intraday history used by + pricing(). + """ pair = SecurityPair.__new__(SecurityPair) - pair.base_yf = FakeYFinanceSecurity("BASE", base_info, base_history) + pair.base_yf = FakeYFinanceSecurity("BASE", base_info, base_history, base_daily) pair.underlying_yf = FakeYFinanceSecurity( - "UNDER", underlying_info, underlying_history + "UNDER", underlying_info, underlying_history, underlying_daily ) pair.calendar = FakeMarketCalendar(base_open, underlying_open, close_time, tz) + # Inject fake FX security when provided — prevents any network calls in FX path + pair.ccy_pair_yf = ( + FakeYFinanceSecurity("FX", {}, fx_history, fx_daily) + if fx_history is not None + else None + ) return pair diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py new file mode 100644 index 0000000..77d155b --- /dev/null +++ b/tests/test_benchmark.py @@ -0,0 +1,137 @@ +"""Tests for the benchmark module.""" + +import numpy as np +import pandas as pd +import pytest + +from afterquote._benchmark import benchmark, metrics +from tests.conftest import make_ohlc, make_security_pair + + +def _make_daily_ohlc(closes, opens=None, start="2026-05-01"): + """Build a daily OHLC DataFrame from a list of close prices.""" + n = len(closes) + opens = opens or closes + rows = [(o, c * 1.01, c * 0.99, c) for o, c in zip(opens, closes)] + index = pd.bdate_range(start=start, periods=n) + return pd.DataFrame( + { + "Open": [r[0] for r in rows], + "High": [r[1] for r in rows], + "Low": [r[2] for r in rows], + "Close": [r[3] for r in rows], + }, + index=index, + ) + + +def _make_benchmark_pair(base_closes, und_closes, leverage=1, fx_closes=None): + """Wire up a pair with daily history suitable for benchmark().""" + + base_daily = _make_daily_ohlc(base_closes) + und_daily = _make_daily_ohlc(und_closes) + + base_info = { + "longName": "Test Base", + "leverage": leverage, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + "currency": "USD", + } + und_info = { + "longName": "Test Und", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + "currency": "USD", + } + + dummy_close = make_ohlc( + [(100.0, 100.0, 100.0, 100.0)], start="2026-05-01", tz="Europe/London" + ) + return make_security_pair( + base_info, + dummy_close, + und_info, + dummy_close, + close_time=pd.Timestamp("2026-05-01 16:30:00+01:00"), + base_daily=base_daily, + underlying_daily=und_daily, + fx_daily=(_make_daily_ohlc(fx_closes) if fx_closes is not None else None), + ) + + +class TestBenchmarkOutput: + def test_returns_dataframe_with_expected_columns(self): + pair = _make_benchmark_pair( + base_closes=[100.0] * 6, + und_closes=[100.0, 101.0, 102.0, 101.0, 103.0, 104.0], + ) + results = benchmark(pair, days=3) + assert set(results.columns) == { + "base_close", + "synth_open", + "actual_open", + "residual", + "direction_correct", + } + + def test_respects_days_limit(self): + pair = _make_benchmark_pair( + base_closes=[100.0] * 10, + und_closes=[100.0 + i for i in range(10)], + ) + results = benchmark(pair, days=5) + assert len(results) <= 5 + + def test_flat_underlying_synth_open_equals_base_close(self): + # Underlying flat every day → synth_open == base_close (no move) + pair = _make_benchmark_pair( + base_closes=[200.0, 200.0, 200.0, 200.0, 200.0], + und_closes=[100.0, 100.0, 100.0, 100.0, 100.0], + leverage=1, + ) + results = benchmark(pair, days=3) + assert (results["synth_open"] == results["base_close"]).all() + + def test_direction_correct_flag(self): + pair = _make_benchmark_pair( + base_closes=[100.0, 110.0, 105.0, 108.0, 112.0], + und_closes=[100.0, 105.0, 103.0, 106.0, 110.0], + leverage=1, + ) + results = benchmark(pair, days=3) + assert "direction_correct" in results.columns + assert results["direction_correct"].dtype == bool + + +class TestMetrics: + def _make_results(self, residuals, directions): + return pd.DataFrame( + { + "base_close": [100.0] * len(residuals), + "synth_open": [100.0 + r for r in residuals], + "actual_open": [100.0] * len(residuals), + "residual": residuals, + "direction_correct": directions, + } + ) + + def test_rmse(self): + results = self._make_results([3.0, -4.0], [True, False]) + m = metrics(results) + assert m["rmse"] == pytest.approx(np.sqrt((9 + 16) / 2)) + + def test_mae(self): + results = self._make_results([3.0, -5.0], [True, False]) + m = metrics(results) + assert m["mae"] == pytest.approx(4.0) + + def test_hit_rate(self): + results = self._make_results([1.0, -1.0, 1.0], [True, False, True]) + m = metrics(results) + assert m["direction_correct"] == pytest.approx(2 / 3) + + def test_n(self): + results = self._make_results([1.0, 2.0, 3.0], [True, True, True]) + assert metrics(results)["n"] == 3 diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..8912dce --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,135 @@ +"""Tests for the CLI entrypoint.""" + +import pandas as pd +import pytest + +from afterquote._cli import main +from tests.conftest import make_ohlc, make_security_pair + + +def _make_pair(base_open=False): + underlying = make_ohlc([(100.0, 101.0, 99.5, 101.0)]) + base_close = make_ohlc( + [(200.0, 200.5, 199.5, 200.0)], start="2026-06-18 16:30", tz="Europe/London" + ) + info = { + "longName": "2x Lev", + "leverage": 2, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + "currency": "USD", + } + und_info = { + "longName": "Under ETF", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + "currency": "USD", + } + # Daily data for benchmark + confidence paths + daily_closes = [100.0 * (1.01**i) for i in range(100)] + idx = pd.bdate_range("2026-01-01", periods=100) + daily = pd.DataFrame( + { + "Open": daily_closes, + "High": [c * 1.01 for c in daily_closes], + "Low": [c * 0.99 for c in daily_closes], + "Close": daily_closes, + }, + index=idx, + ) + return make_security_pair( + info, + base_close, + und_info, + underlying, + base_open=base_open, + close_time=pd.Timestamp("2026-06-18 16:30:00+01:00"), + base_daily=daily, + underlying_daily=daily.copy(), + ) + + +class TestCLIInfo: + def test_info_prints_output(self, capsys, monkeypatch): + pair = _make_pair() + monkeypatch.setattr("afterquote._cli.SecurityPair", lambda b, u: pair) + main(["BASE", "UNDER"]) + out = capsys.readouterr().out + assert "base_is_live" in out + assert "quote_price" in out + + def test_pricing_flag_prints_impl_columns(self, capsys, monkeypatch): + pair = _make_pair() + monkeypatch.setattr("afterquote._cli.SecurityPair", lambda b, u: pair) + main(["BASE", "UNDER", "--pricing"]) + out = capsys.readouterr().out + assert "Impl_Open" in out + assert "Impl_Close" in out + + def test_as_of_flag_parsed(self, capsys, monkeypatch): + pair = _make_pair() + monkeypatch.setattr("afterquote._cli.SecurityPair", lambda b, u: pair) + main(["BASE", "UNDER", "--as-of", "2026-06-18 14:00:00-04:00"]) + out = capsys.readouterr().out + assert len(out) > 0 + + def test_bad_ticker_exits_1(self, monkeypatch): + monkeypatch.setattr( + "afterquote._cli.SecurityPair", + lambda b, u: (_ for _ in ()).throw(ValueError("bad ticker")), + ) + with pytest.raises(SystemExit) as exc: + main(["BAD", "TICKER"]) + assert exc.value.code == 1 + + def test_confidence_flag_prints_band(self, capsys, monkeypatch): + pair = _make_pair() + monkeypatch.setattr("afterquote._cli.SecurityPair", lambda b, u: pair) + main(["BASE", "UNDER", "--confidence", "0.95"]) + out = capsys.readouterr().out + assert "lower_bound" in out + assert "upper_bound" in out + + def test_benchmark_flag_prints_metrics(self, capsys, monkeypatch): + pair = _make_pair() + monkeypatch.setattr("afterquote._cli.SecurityPair", lambda b, u: pair) + main(["BASE", "UNDER", "--benchmark"]) + out = capsys.readouterr().out + assert "synth_open" in out + assert "residual" in out + assert "rmse:" in out + assert "mae:" in out + assert "tracking_error:" in out + + def test_correlation_flag_prints_value(self, capsys, monkeypatch): + pair = _make_pair() + monkeypatch.setattr("afterquote._cli.SecurityPair", lambda b, u: pair) + main(["BASE", "UNDER", "--correlation"]) + out = capsys.readouterr().out + assert "correlation:" in out + + def test_holdings_flag_prints_pnl(self, capsys, monkeypatch, tmp_path): + import pandas as pd + + csv = tmp_path / "h.csv" + pd.DataFrame([{"base": "A", "underlying": "B", "quantity": 10}]).to_csv( + csv, index=False + ) + monkeypatch.setattr( + "afterquote._cli.portfolio_pnl", + lambda path, as_of=None: pd.DataFrame( + [{"base": "A", "underlying": "B", "quantity": 10, "pnl": 50.0}] + ), + ) + main(["BASE", "UNDER", "--holdings", str(csv)]) + out = capsys.readouterr().out + assert "pnl" in out + + def test_benchmark_and_pricing_mutually_exclusive(self): + with pytest.raises(SystemExit): + main(["BASE", "UNDER", "--benchmark", "--pricing"]) + + def test_correlation_and_pricing_mutually_exclusive(self): + with pytest.raises(SystemExit): + main(["BASE", "UNDER", "--correlation", "--pricing"]) diff --git a/tests/test_confidence.py b/tests/test_confidence.py new file mode 100644 index 0000000..55d13fc --- /dev/null +++ b/tests/test_confidence.py @@ -0,0 +1,231 @@ +"""Tests for the confidence band on QuoteInfo.""" + +import pandas as pd +import pytest + +from tests.conftest import make_ohlc, make_security_pair + + +def _daily_ohlc(closes, opens=None, start="2026-04-01"): + """Build a tz-naive daily OHLC DataFrame from close prices.""" + n = len(closes) + opens = opens if opens is not None else closes + index = pd.bdate_range(start=start, periods=n) + rows = [(o, c * 1.01, c * 0.99, c) for o, c in zip(opens, closes)] + return pd.DataFrame( + { + "Open": [r[0] for r in rows], + "High": [r[1] for r in rows], + "Low": [r[2] for r in rows], + "Close": [r[3] for r in rows], + }, + index=index, + ) + + +def _confidence_pair(base_closes, base_opens, und_closes, leverage=1): + """Build a pair where intraday underlying is flat (pricing() → quote=anchor) + and daily history is independently set for benchmark(). + + Flat intraday → quote_price always equals the base_close anchor. Daily base + closes/opens are independent → residuals are exactly controlled. + """ + base_close_bar = make_ohlc( + [(200.0, 200.5, 199.5, 200.0)], + start="2026-06-18 16:30", + tz="Europe/London", + ) + flat_intraday = make_ohlc([(100.0, 100.5, 99.5, 100.0)] * 3, freq="1min") + base_daily = _daily_ohlc(base_closes, opens=base_opens) + und_daily = _daily_ohlc(und_closes) + + return make_security_pair( + base_info={ + "longName": "Test Base", + "leverage": leverage, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + "currency": "USD", + }, + base_history=base_close_bar, + underlying_info={ + "longName": "Test Und", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + "currency": "USD", + }, + underlying_history=flat_intraday, + close_time=pd.Timestamp("2026-06-18 16:30:00+01:00"), + base_daily=base_daily, + underlying_daily=und_daily, + ) + + +class TestConfidenceBand: + def test_no_confidence_no_band_columns(self): + # baseline: lower_bound/upper_bound absent without confidence= + pair = _confidence_pair( + base_closes=[100.0] * 5, + base_opens=[100.0, 99.0, 101.0, 98.0, 102.0], + und_closes=[100.0] * 5, + ) + info = pair.info() + assert "lower_bound" not in info.columns + assert "upper_bound" not in info.columns + assert "quote_price" in info.columns + + def test_confidence_attaches_band(self): + pair = _confidence_pair( + base_closes=[100.0] * 5, + base_opens=[100.0, 99.0, 101.0, 98.0, 102.0], + und_closes=[100.0] * 5, + ) + info = pair.info(confidence=0.95) + assert "lower_bound" in info.columns + assert "upper_bound" in info.columns + assert "quote_price" in info.columns + + def test_known_residuals_exact_band(self): + # Flat underlying → synth_open = base_close = 100. + # base_opens[i+1] = 99, 101, 98, 102 → residuals = [1, -1, 2, -2]. + # quote_price = 200 (anchor). np.percentile([-2,-1,1,2], [2.5,97.5]) + # -> [-1.925, 1.925] by linear interpolation. + pair = _confidence_pair( + base_closes=[100.0] * 5, + base_opens=[100.0, 99.0, 101.0, 98.0, 102.0], + und_closes=[100.0] * 5, + ) + info = pair.info(confidence=0.95) + quote_price = info["quote_price"].iloc[0] + assert quote_price == pytest.approx(200.0) + assert info["lower_bound"].iloc[0] == pytest.approx(200.0 - 1.925) + assert info["upper_bound"].iloc[0] == pytest.approx(200.0 + 1.925) + + def test_band_asymmetric_for_skewed_residuals(self): + # residuals = [1, -3, 2, -2] (opens: 99, 103, 98, 102). + # sorted = [-3, -2, 1, 2]; 2.5pct at -2.925, 97.5pct at 1.925. + # Asymmetry proves we're not assuming symmetric Gaussian errors. + pair = _confidence_pair( + base_closes=[100.0] * 5, + base_opens=[100.0, 99.0, 103.0, 98.0, 102.0], + und_closes=[100.0] * 5, + ) + info = pair.info(confidence=0.95) + qp = info["quote_price"].iloc[0] + low, high = info["lower_bound"].iloc[0], info["upper_bound"].iloc[0] + assert low == pytest.approx(qp - 2.925) + assert high == pytest.approx(qp + 1.925) + assert abs(qp - low) != pytest.approx(abs(qp - high)) + + def test_band_widens_with_higher_confidence(self): + # 99% band uses [0.5, 99.5] percentiles -> strictly wider than 95%. + pair = _confidence_pair( + base_closes=[100.0] * 5, + base_opens=[100.0, 99.0, 103.0, 98.0, 102.0], + und_closes=[100.0] * 5, + ) + info95 = pair.info(confidence=0.95) + info99 = pair.info(confidence=0.99) + w95 = info95["upper_bound"].iloc[0] - info95["lower_bound"].iloc[0] + w99 = info99["upper_bound"].iloc[0] - info99["lower_bound"].iloc[0] + assert w99 > w95 + + +class TestBandGates: + """Band is only attached on the synthetic path.""" + + def test_live_path_no_band_even_with_confidence(self): + base_close_bar = make_ohlc( + [(200.0, 200.5, 199.5, 200.0)], + start="2026-06-18 16:30", + tz="Europe/London", + ) + pair = make_security_pair( + base_info={ + "longName": "Live", + "leverage": 2, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + }, + base_history=base_close_bar, + underlying_info={ + "longName": "Under", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + }, + underlying_history=base_close_bar, + base_open=True, + close_time=pd.Timestamp("2026-06-18 16:30:00+01:00"), + ) + info = pair.info(confidence=0.95) + assert bool(info["base_is_live"].iloc[0]) + assert "lower_bound" not in info.columns + assert "upper_bound" not in info.columns + + def test_no_data_path_no_band_even_with_confidence(self): + empty = pd.DataFrame(columns=["Open", "High", "Low", "Close"]) + base_close_bar = make_ohlc( + [(200.0, 200.5, 199.5, 200.0)], + start="2026-06-18 16:30", + tz="Europe/London", + ) + pair = make_security_pair( + base_info={ + "longName": "X", + "leverage": 2, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + }, + base_history=base_close_bar, + underlying_info={ + "longName": "Y", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + }, + underlying_history=empty, + close_time=pd.Timestamp("2026-06-18 16:30:00+01:00"), + ) + info = pair.info(confidence=0.95) + assert "lower_bound" not in info.columns + assert "upper_bound" not in info.columns + + def test_empty_residuals_raises(self): + # Empty daily underlying -> every daily_return is NaN -> residuals all + # NaN -> dropna empty -> _confidence_band raises ValueError. + empty_daily = pd.DataFrame( + columns=["Open", "High", "Low", "Close"], + index=pd.DatetimeIndex([], dtype="datetime64[ns]"), + ) + base_close_bar = make_ohlc( + [(200.0, 200.5, 199.5, 200.0)], + start="2026-06-18 16:30", + tz="Europe/London", + ) + flat_intraday = make_ohlc([(100.0, 100.5, 99.5, 100.0)] * 3, freq="1min") + base_daily = _daily_ohlc([100.0] * 5, opens=[100.0] * 5) + pair = make_security_pair( + base_info={ + "longName": "X", + "leverage": 1, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + "currency": "USD", + }, + base_history=base_close_bar, + underlying_info={ + "longName": "Y", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + "currency": "USD", + }, + underlying_history=flat_intraday, + base_daily=base_daily, + underlying_daily=empty_daily, + close_time=pd.Timestamp("2026-06-18 16:30:00+01:00"), + ) + with pytest.raises(ValueError, match="no.*valid residual|benchmark"): + pair.info(confidence=0.95) diff --git a/tests/test_correlation.py b/tests/test_correlation.py new file mode 100644 index 0000000..a2b5592 --- /dev/null +++ b/tests/test_correlation.py @@ -0,0 +1,92 @@ +"""Tests for SecurityPair.correlation().""" + +import random +import warnings + +import pandas as pd +import pytest + +from tests.conftest import make_ohlc, make_security_pair + + +def _daily_ohlc(closes, start="2026-01-01"): + idx = pd.bdate_range(start=start, periods=len(closes)) + return pd.DataFrame( + { + "Open": closes, + "High": [c * 1.01 for c in closes], + "Low": [c * 0.99 for c in closes], + "Close": closes, + }, + index=idx, + ) + + +def _make_corr_pair(base_closes, und_closes, leverage=1): + close_bar = make_ohlc( + [(200.0, 200.5, 199.5, 200.0)], + start="2026-06-18 16:30", + tz="Europe/London", + ) + return make_security_pair( + { + "longName": "Base", + "leverage": leverage, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + "currency": "USD", + }, + close_bar, + { + "longName": "Und", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + "currency": "USD", + }, + close_bar, + close_time=pd.Timestamp("2026-06-18 16:30:00+01:00"), + base_daily=_daily_ohlc(base_closes), + underlying_daily=_daily_ohlc(und_closes), + ) + + +class TestCorrelation: + def test_correlated_pair_returns_high_corr_no_warning(self): + closes = [100.0 * (1.01**i) for i in range(100)] + pair = _make_corr_pair(closes, closes) + with warnings.catch_warnings(): + warnings.simplefilter("error") + corr = pair.correlation(days=90, warn=True) + assert corr == pytest.approx(1.0, abs=1e-6) + + def test_uncorrelated_pair_warns(self): + random.seed(42) + base = [100.0 + i * 0.1 for i in range(100)] + und = [100.0 + random.uniform(-1, 1) for _ in range(100)] + pair = _make_corr_pair(base, und) + with pytest.warns(UserWarning, match="Low correlation"): + corr = pair.correlation(days=90, warn=True) + assert abs(corr) < 0.5 + + def test_warn_false_suppresses_warning(self): + random.seed(42) + base = [100.0 + i * 0.1 for i in range(100)] + und = [100.0 + random.uniform(-1, 1) for _ in range(100)] + pair = _make_corr_pair(base, und) + with warnings.catch_warnings(): + warnings.simplefilter("error") + corr = pair.correlation(days=90, warn=False) + assert abs(corr) < 0.5 + + def test_short_history_returns_nan(self): + closes = [100.0] + pair = _make_corr_pair(closes, closes) + corr = pair.correlation(days=90, warn=False) + assert pd.isna(corr) + + def test_flat_series_returns_nan(self): + closes = [100.0] * 100 + pair = _make_corr_pair(closes, closes) + corr = pair.correlation(days=90, warn=False) + assert pd.isna(corr) diff --git a/tests/test_fx.py b/tests/test_fx.py new file mode 100644 index 0000000..bf830b5 --- /dev/null +++ b/tests/test_fx.py @@ -0,0 +1,146 @@ +"""Tests for FX currency adjustment in synthetic pricing.""" + +import pandas as pd +import pytest + +from tests.conftest import make_ohlc, make_security_pair + +BASE_INFO_GBP = { + "longName": "3x Tesla GBp ETP", + "leverage": 3, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + "currency": "GBp", # pence — normalised to GBP in get_currency() +} +UNDERLYING_INFO_USD = { + "longName": "Tesla Inc", + "leverage": 1, + "exchange": "NMS", + "timeZoneFullName": "America/New_York", + "currency": "USD", +} +BASE_INFO_USD = { + "longName": "3x S&P ETP", + "leverage": 3, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + "currency": "USD", +} + +_CLOSE_TIME = pd.Timestamp("2026-06-18 16:30:00+01:00") + + +def _base_close(anchor=100.0): + return make_ohlc( + [(anchor, anchor, anchor, anchor)], + start="2026-06-18 16:30", + tz="Europe/London", + ) + + +class TestCurrencyNormalisation: + """GBp (pence) is normalised to GBP when building the FX ticker.""" + + def test_get_currency_uppercase(self): + from tests.conftest import FakeYFinanceSecurity + + sec = FakeYFinanceSecurity("X", {"currency": "GBp"}, pd.DataFrame()) + assert sec.get_currency() == "GBP" + + def test_get_currency_already_upper(self): + from tests.conftest import FakeYFinanceSecurity + + sec = FakeYFinanceSecurity("X", {"currency": "USD"}, pd.DataFrame()) + assert sec.get_currency() == "USD" + + +class TestSameCurrencyNoFxLeg: + """When base and underlying share a currency, FX leg is a no-op (factor = 1).""" + + def test_same_currency_result_unchanged(self): + # USD/USD pair — FX should not alter the math + underlying = make_ohlc([(100.0, 101.0, 99.0, 101.0)]) + pair = make_security_pair( + BASE_INFO_USD, + _base_close(200.0), + UNDERLYING_INFO_USD, + underlying, + close_time=_CLOSE_TIME, + ) + pricing = pair.pricing() + # leverage=3, underlying +1%, anchor=200 → 200*(1+3*0.01) = 206 + assert pricing["Impl_Close"].iloc[0] == pytest.approx(206.0) + + +class TestCrossCurrencyFxApplied: + """FX leg multiplies into the return when currencies differ.""" + + def _make_cross_pair(self, underlying_rows, fx_rows, anchor=100.0): + underlying = make_ohlc(underlying_rows) + fx = make_ohlc(fx_rows) + return make_security_pair( + BASE_INFO_GBP, + _base_close(anchor), + UNDERLYING_INFO_USD, + underlying, + close_time=_CLOSE_TIME, + fx_history=fx, + ) + + def test_flat_underlying_fx_move_applies(self): + # Underlying flat (0%), FX +1% (USD strengthens vs GBP) + # leverage=3, but FX is 1x: Impl_Close = 100 * 1.0 (underlying) * 1.01 (FX) = 101 + pair = self._make_cross_pair( + underlying_rows=[(100.0, 100.0, 100.0, 100.0)], + fx_rows=[(1.0, 1.01, 1.0, 1.01)], + ) + pricing = pair.pricing() + assert pricing["Impl_Close"].iloc[0] == pytest.approx(101.0) + + def test_underlying_move_leveraged_fx_not(self): + # Underlying +1%, FX +1%, leverage=3 + # Expected: anchor * (1 + 3*0.01) * 1.01 = 100 * 1.03 * 1.01 = 104.03 + pair = self._make_cross_pair( + underlying_rows=[(100.0, 101.0, 99.0, 101.0)], + fx_rows=[(1.0, 1.01, 1.0, 1.01)], + ) + pricing = pair.pricing() + assert pricing["Impl_Close"].iloc[0] == pytest.approx(100 * 1.03 * 1.01) + + def test_fx_weakening_base_reduces_return(self): + # Underlying +1%, FX -1% (USD weakens vs GBP): GBP return is reduced + # Expected: 100 * 1.03 * 0.99 = 101.97 + pair = self._make_cross_pair( + underlying_rows=[(100.0, 101.0, 99.0, 101.0)], + fx_rows=[(1.0, 1.0, 0.99, 0.99)], + ) + pricing = pair.pricing() + assert pricing["Impl_Close"].iloc[0] == pytest.approx(100 * 1.03 * 0.99) + + def test_open_is_anchor_when_no_gap(self): + # First bar open should still be the anchor (no gap on bar 0) + pair = self._make_cross_pair( + underlying_rows=[(100.0, 101.0, 99.0, 101.0)], + fx_rows=[(1.0, 1.01, 1.0, 1.01)], + ) + pricing = pair.pricing() + assert pricing["Impl_Open"].iloc[0] == pytest.approx(100.0) + + def test_high_low_include_fx(self): + # Underlying High=+2%, Low=-1%, FX +1%, leverage=1 + # Impl_High = anchor * (1 + 1*0.02) * 1.01 = 102 * 1.01 = 103.02 + # Impl_Low = anchor * (1 + 1*(-0.01)) * 1.01 = 99 * 1.01 = 99.99 + underlying = make_ohlc([(100.0, 102.0, 99.0, 100.0)]) + fx = make_ohlc([(1.0, 1.01, 1.0, 1.01)]) + base_info_1x = {**BASE_INFO_GBP, "leverage": 1} + pair = make_security_pair( + base_info_1x, + _base_close(100.0), + UNDERLYING_INFO_USD, + underlying, + close_time=_CLOSE_TIME, + fx_history=fx, + ) + pricing = pair.pricing() + assert pricing["Impl_High"].iloc[0] == pytest.approx(102.0 * 1.01) + assert pricing["Impl_Low"].iloc[0] == pytest.approx(99.0 * 1.01) diff --git a/tests/test_holdings.py b/tests/test_holdings.py new file mode 100644 index 0000000..48832a9 --- /dev/null +++ b/tests/test_holdings.py @@ -0,0 +1,133 @@ +"""Tests for portfolio holdings P&L ingestion.""" + +import json + +import pandas as pd +import pytest + +from afterquote._holdings import portfolio_pnl +from tests.conftest import make_ohlc, make_security_pair + + +def _make_fake_pair(quote_price, close_price, anchor=None): + """Returns a factory callable that produces a stubbed SecurityPair.""" + underlying = make_ohlc([(100.0, 101.0, 99.0, 101.0)]) + close = make_ohlc( + [(close_price, close_price, close_price, close_price)], + start="2026-06-18 16:30", + tz="Europe/London", + ) + # Underlying that moves to produce the desired quote_price + move = (quote_price - close_price) / close_price + underlying = make_ohlc([(100.0, 101.0, 99.0, 100.0 * (1 + move))]) + info = { + "longName": "Fake Base", + "leverage": 1, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + "currency": "USD", + } + und_info = { + "longName": "Fake Und", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + "currency": "USD", + } + return make_security_pair( + info, + close, + und_info, + underlying, + close_time=pd.Timestamp("2026-06-18 16:30:00+01:00"), + ) + + +def _write_csv(tmp_path, rows): + p = tmp_path / "positions.csv" + pd.DataFrame(rows).to_csv(p, index=False) + return p + + +def _write_json(tmp_path, rows): + p = tmp_path / "positions.json" + p.write_text(json.dumps(rows)) + return p + + +class TestPortfolioPnl: + def test_single_position_pnl(self, tmp_path): + pair = _make_fake_pair(quote_price=110.0, close_price=100.0) + path = _write_csv(tmp_path, [{"base": "A", "underlying": "B", "quantity": 10}]) + result = portfolio_pnl(path, pair_factory=lambda b, u: pair) + pos_row = result[result["base"] == "A"].iloc[0] + assert pos_row["pnl"] == pytest.approx(10 * (110.0 - 100.0)) + + def test_total_row_sums_pnl(self, tmp_path): + pair1 = _make_fake_pair(110.0, 100.0) + pair2 = _make_fake_pair(95.0, 100.0) + pairs = {"A": pair1, "B": pair2} + path = _write_csv( + tmp_path, + [ + {"base": "A", "underlying": "X", "quantity": 10}, + {"base": "B", "underlying": "Y", "quantity": 5}, + ], + ) + result = portfolio_pnl(path, pair_factory=lambda b, u: pairs[b]) + total = result[result["base"] == "TOTAL"].iloc[0] + # A: 10*(110-100)=100, B: 5*(95-100)=-25, total=75 + assert total["pnl"] == pytest.approx(75.0) + + def test_json_input(self, tmp_path): + pair = _make_fake_pair(110.0, 100.0) + path = _write_json(tmp_path, [{"base": "A", "underlying": "B", "quantity": 2}]) + result = portfolio_pnl(path, pair_factory=lambda b, u: pair) + assert len(result) == 2 # 1 position + total row + + def test_missing_columns_raises(self, tmp_path): + path = _write_csv(tmp_path, [{"base": "A", "quantity": 10}]) + with pytest.raises(ValueError, match="missing columns"): + portfolio_pnl(path, pair_factory=lambda b, u: None) + + def test_result_has_total_row(self, tmp_path): + pair = _make_fake_pair(105.0, 100.0) + path = _write_csv(tmp_path, [{"base": "A", "underlying": "B", "quantity": 1}]) + result = portfolio_pnl(path, pair_factory=lambda b, u: pair) + assert "TOTAL" in result["base"].values + + def test_as_of_passed_through(self, tmp_path): + """as_of is forwarded to each pair's info() call.""" + calls = [] + + class SpyPair: + def info(self, as_of=None): + calls.append(as_of) + return make_security_pair( + { + "longName": "X", + "leverage": 1, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + "currency": "USD", + }, + make_ohlc( + [(100.0, 100.0, 100.0, 100.0)], + start="2026-06-18 16:30", + tz="Europe/London", + ), + { + "longName": "Y", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + "currency": "USD", + }, + make_ohlc([(100.0, 100.0, 100.0, 100.0)]), + close_time=pd.Timestamp("2026-06-18 16:30:00+01:00"), + ).info() + + path = _write_csv(tmp_path, [{"base": "A", "underlying": "B", "quantity": 1}]) + as_of = pd.Timestamp("2026-06-18 14:00:00-04:00") + portfolio_pnl(path, as_of=as_of, pair_factory=lambda b, u: SpyPair()) + assert calls[0] == as_of diff --git a/tests/test_info.py b/tests/test_info.py index 5ab75fd..c4a7183 100644 --- a/tests/test_info.py +++ b/tests/test_info.py @@ -49,6 +49,47 @@ def test_quote_price_is_last_impl_close(self, simple_pair): assert info["quote_price"].iloc[0] == pricing["Impl_Close"].iloc[-1] +class TestInfoAsOf: + """as_of= parameter flows through info() correctly.""" + + def test_as_of_synthetic_has_adj_return(self, simple_pair): + as_of = pd.Timestamp("2026-06-18 19:00:00+01:00") + info = simple_pair.info(as_of=as_of) + assert not bool(info["base_is_live"].iloc[0]) + assert "adj_percent_return" in info.columns + + def test_as_of_quote_price_is_last_impl_close(self, simple_pair): + as_of = pd.Timestamp("2026-06-18 19:00:00+01:00") + info = simple_pair.info(as_of=as_of) + pricing = simple_pair.pricing(as_of=as_of) + assert info["quote_price"].iloc[0] == pricing["Impl_Close"].iloc[-1] + + def test_as_of_live_base_returns_live_info(self): + base_close = make_ohlc( + [(200.0, 200.5, 199.5, 200.0)], start="2026-06-18 16:30", tz="Europe/London" + ) + pair = make_security_pair( + { + "longName": "2x Lev", + "leverage": 2, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + }, + base_close, + { + "longName": "Under", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + }, + base_close, + base_open=True, + close_time=pd.Timestamp("2026-06-18 16:30:00+01:00"), + ) + info = pair.info(as_of=pd.Timestamp("2026-06-18 13:00:00+01:00")) + assert bool(info["base_is_live"].iloc[0]) + + class TestInfoNoData: """When underlying has no data after base close, return base close as quote.""" diff --git a/tests/test_pricing.py b/tests/test_pricing.py index 50ae4df..6c69b58 100644 --- a/tests/test_pricing.py +++ b/tests/test_pricing.py @@ -141,3 +141,51 @@ def test_columns(self, simple_pair): "Impl_Low", "Impl_Close", ] + + +class TestPricingAsOf: + """as_of= parameter flows through without altering the math (fakes don't filter).""" + + def test_as_of_returns_correct_columns(self, simple_pair): + as_of = pd.Timestamp("2026-06-18 19:00:00+01:00") + pricing = simple_pair.pricing(as_of=as_of) + assert list(pricing.columns) == [ + "Impl_Open", + "Impl_High", + "Impl_Low", + "Impl_Close", + ] + + def test_as_of_preserves_anchor_and_leverage(self, simple_pair): + as_of = pd.Timestamp("2026-06-18 19:00:00+01:00") + pricing = simple_pair.pricing(as_of=as_of) + # anchor=200, underlying +1%, leverage=2 → first close = 204 + assert pricing["Impl_Close"].iloc[0] == 204.0 + + def test_as_of_raises_when_base_is_live(self): + import pytest + from tests.conftest import make_ohlc, make_security_pair + + base_close = make_ohlc( + [(200.0, 200.5, 199.5, 200.0)], start="2026-06-18 16:30", tz="Europe/London" + ) + pair = make_security_pair( + { + "longName": "2x Lev", + "leverage": 2, + "exchange": "LSE", + "timeZoneFullName": "Europe/London", + }, + base_close, + { + "longName": "Under", + "leverage": 1, + "exchange": "PCX", + "timeZoneFullName": "America/New_York", + }, + base_close, + base_open=True, + close_time=pd.Timestamp("2026-06-18 16:30:00+01:00"), + ) + with pytest.raises(RuntimeError): + pair.pricing(as_of=pd.Timestamp("2026-06-18 17:00:00+01:00"))