dodox/smaug
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: cap-tier filtering, Alpaca cost model, README cleanup

Browse Source 
- simulate.py: --cap-tier large|mid|small|micro; yfinance market cap fetch
  with DB cache (ticker_meta table); argv fix for main.py dispatch
- plot.py: equity curves now show cap tiers with Alpaca costs (zero commission);
  HP sweep uses Alpaca cost decomposition; SPY line clamped to last strategy date
- db/models.py: TickerMeta table
- db/db.py: get_cached_market_caps, upsert_market_caps
- README: add --cap-tier to simulate docs; backfill note (~3 days for 2 years
  at SEC 10 req/s limit); remove duplicate setup block; remove em-dashes in prose;
  results table tilde estimates to be updated once cap-tier sims complete

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
dodox
2026-05-26 18:10:09 +02:00
co-authored by Claude Sonnet 4.6
parent 56ec0b4a81
commit d0e98b9cb7
6 changed files with 127 additions and 381 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backtest/plot.py
+20 -21
View File
@@ -41,15 +41,11 @@ def plot_hp_heatmap(prices: dict, out_dir: str = PLOTS_DIR) -> str:
hold_days = [3, 5, 7, 10, 14, 21, 30]
rt_pcts = [0.3, 0.5, 0.7, 1.0, 1.2, 1.5, 2.0]
# decompose round-trip into (spread, slippage, commission) that sum correctly:
# roundtrip = 2*spread + slippage + 2*commission
# allocate 40% spread, 40% slippage, 20% commission (all relative to RT)
# => spread = RT*0.4/2 = RT*0.2 (one-way)
# => slippage = RT*0.4
# => commission = RT*0.2/2 = RT*0.1 (one-way)
# verify: 2*0.2 + 0.4 + 2*0.1 = 0.4+0.4+0.2 = 1.0 * RT ✓
# Alpaca: zero commission. Decompose RT into spread + slippage only (50/50).
# roundtrip = 2*spread + slippage => spread = RT*0.25, slippage = RT*0.5
# verify: 2*0.25 + 0.5 = 1.0 * RT ✓
def _costs(rt):
return dict(spread=rt * 0.2, slippage=rt * 0.4, commission=rt * 0.1)
return dict(spread=rt * 0.25, slippage=rt * 0.5, commission=0)
rows_excess = []
rows_ann = []
@@ -116,7 +112,7 @@ def plot_hp_heatmap(prices: dict, out_dir: str = PLOTS_DIR) -> str:
ax.text(j, i, txt, ha="center", va="center", fontsize=7.5, color=color)
fig.suptitle(
"HP sweep: 1-day entry delay, 10% position size, buy filter only",
"HP sweep: Alpaca (zero commission), 1-day entry delay, 10% position size, all cap tiers",
fontsize=12,
)
plt.tight_layout()
@@ -135,22 +131,25 @@ def plot_equity_curves(prices: dict, out_dir: str = PLOTS_DIR) -> str:
"""
matplotlib, plt, mdates, np = _get_matplotlib()
# Alpaca zero-commission costs by cap tier (spread + slippage only)
scenarios = [
{"label": "0% RT cost (theoretical)", "spread": 0, "slippage": 0, "commission": 0},
{"label": "0.67% RT (best case)", "spread": 0.0014, "slippage": 0.0027, "commission": 0.0007},
{"label": "1.0% RT (mid)", "spread": 0.002, "slippage": 0.004, "commission": 0.001},
{"label": "1.5% RT (realistic small-cap)","spread": 0.003, "slippage": 0.006, "commission": 0.0015},
{"label": "Large cap (~0.2% RT)", "cap_tier": "large", "spread": 0.001, "slippage": 0.001},
{"label": "Mid cap (~0.5% RT)", "cap_tier": "mid", "spread": 0.0025, "slippage": 0.0025},
{"label": "Small cap (~0.8% RT)", "cap_tier": "small", "spread": 0.004, "slippage": 0.004},
{"label": "All tickers (0% RT)", "cap_tier": None, "spread": 0, "slippage": 0},
]
fig, ax = plt.subplots(figsize=(13, 7))
colors = ["#2ecc71", "#3498db", "#e67e22", "#e74c3c"]
sim_start = sim_end = None
colors = ["#2ecc71", "#3498db", "#e67e22", "#aaaaaa"]
sim_start = None
last_curve_date = None
for sc, color in zip(scenarios, colors):
s = Strategy(
holding_days=7, buy_delay=1,
spread=sc["spread"], slippage=sc["slippage"], commission=sc["commission"],
spread=sc["spread"], slippage=sc["slippage"], commission=0,
cap_tier=sc["cap_tier"],
)
r = simulate(s, prices=prices)
curve = r.get("equity_curve", [])
@@ -158,7 +157,7 @@ def plot_equity_curves(prices: dict, out_dir: str = PLOTS_DIR) -> str:
continue
sim_start = sim_start or r["period"]["start"]
sim_end = r["period"]["end"]
last_curve_date = curve[-1][0] # actual last signal date in this curve
dates = [datetime.strptime(d, "%Y-%m-%d") for d, _ in curve]
values = [v for _, v in curve]
@@ -166,10 +165,10 @@ def plot_equity_curves(prices: dict, out_dir: str = PLOTS_DIR) -> str:
ax.plot(dates, [v / base * 100 for v in values],
label=sc["label"], color=color, linewidth=1.8)
# SPY buy-and-hold overlay
# SPY buy-and-hold overlay — clamp to last data point of strategy curves
spy_px = prices.get("SPY", {})
if spy_px and sim_start and sim_end:
spy_dates = sorted(d for d in spy_px if sim_start <= d <= sim_end)
if spy_px and sim_start and last_curve_date:
spy_dates = sorted(d for d in spy_px if sim_start <= d <= last_curve_date)
if spy_dates:
base = spy_px[spy_dates[0]]
ax.plot(
@@ -182,7 +181,7 @@ def plot_equity_curves(prices: dict, out_dir: str = PLOTS_DIR) -> str:
ax.set_xlabel("Date", fontsize=11)
ax.set_ylabel("Portfolio value (indexed to 100)", fontsize=11)
ax.set_title(
"Insider Copytrade: equity curves vs SPY (7d hold, 1d delay, 10% position size)",
"Insider Copytrade: equity curves by cap tier, Alpaca costs (7d hold, 1d delay, 10% position size)",
fontsize=12,
)
ax.legend(fontsize=10)
backtest/simulate.py
+61 -3
View File
@@ -32,7 +32,39 @@ from datetime import datetime, timedelta
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
import config
from db.db import get_signals_for_backtest
from db.db import get_signals_for_backtest, get_cached_market_caps, upsert_market_caps
CAP_TIERS = {
"large": (10_000_000_000, None),
"mid": (2_000_000_000, 10_000_000_000),
"small": (300_000_000, 2_000_000_000),
"micro": (0, 300_000_000),
}
def _fetch_market_caps(tickers: list[str]) -> dict[str, float]:
"""Return market cap for each ticker, using DB cache then yfinance for misses."""
import yfinance as yf
cached = get_cached_market_caps(tickers)
missing = [t for t in tickers if t not in cached]
if missing:
logger.info(f"Fetching market caps for {len(missing)} tickers via yfinance...")
fetched = {}
for ticker in missing:
try:
info = yf.Ticker(ticker).fast_info
cap = getattr(info, "market_cap", None)
if cap:
fetched[ticker] = float(cap)
except Exception:
pass
if fetched:
upsert_market_caps(fetched)
cached.update(fetched)
return cached
logger = logging.getLogger(__name__)
@@ -92,6 +124,7 @@ class Strategy:
spread: float = 0.003,
slippage: float = 0.002,
commission: float = 0.001,
cap_tier: str = None,
):
self.holding_days = holding_days
self.buy_delay = buy_delay
@@ -102,6 +135,7 @@ class Strategy:
self.spread = spread
self.slippage = slippage
self.commission = commission
self.cap_tier = cap_tier # "large" | "mid" | "small" | "micro" | None
# cost applied at entry: half-spread + slippage + commission
@property
@@ -137,6 +171,22 @@ def simulate(strategy: Strategy, prices: dict = None) -> dict:
if not signals:
return {"error": "No signals after filtering"}
if strategy.cap_tier:
tier = CAP_TIERS.get(strategy.cap_tier)
if tier is None:
raise ValueError(f"Unknown cap_tier {strategy.cap_tier!r}. Use: {list(CAP_TIERS)}")
cap_min, cap_max = tier
tickers = list({s["ticker"] for s in signals})
market_caps = _fetch_market_caps(tickers)
signals = [
s for s in signals
if market_caps.get(s["ticker"], 0) >= cap_min
and (cap_max is None or market_caps.get(s["ticker"], 0) < cap_max)
]
logger.info(f"Cap tier '{strategy.cap_tier}': {len(signals)} signals after filtering")
if not signals:
return {"error": f"No signals after cap_tier={strategy.cap_tier} filter"}
if prices is None:
prices = _load_all_prices()
@@ -291,6 +341,7 @@ def simulate(strategy: Strategy, prices: dict = None) -> dict:
"min_score": strategy.min_score,
"min_cluster": strategy.min_cluster,
"roundtrip_cost_pct": round(strategy.roundtrip_cost * 100, 3),
"cap_tier": strategy.cap_tier or "all",
},
"period": {
"start": equity_curve[0][0] if equity_curve else "n/a",
@@ -338,7 +389,7 @@ def _print_results(r: dict):
print(f"{'=' * w}")
print(f" Strategy")
print(f" Hold: {s['holding_days']}d | Delay: {s['buy_delay']}d | Size: {s['position_size']*100:.0f}% of cash")
print(f" Score ≥ {s['min_score']} | Cluster ≥ {s['min_cluster']}")
print(f" Score ≥ {s['min_score']} | Cluster ≥ {s['min_cluster']} | Cap: {s['cap_tier']}")
print(f" Round-trip cost: {s['roundtrip_cost_pct']:.2f}%")
print(f" Period: {period['start']} → {period['end']} ({period['years']}y)")
print(f"{'─' * w}")
@@ -373,6 +424,8 @@ def main():
help="Fraction of available cash per trade (0.10 = 10%%)")
parser.add_argument("--min-score", type=float, default=0.0)
parser.add_argument("--min-cluster", type=int, default=1)
parser.add_argument("--cap-tier", choices=["large", "mid", "small", "micro"],
default=None, help="Filter by market cap tier")
parser.add_argument("--capital", type=float, default=100_000.0)
# Costs
parser.add_argument("--spread", type=float, default=0.003,
@@ -382,7 +435,11 @@ def main():
parser.add_argument("--commission", type=float, default=0.001,
help="Per-trade commission as fraction of notional")
args = parser.parse_args()
# When invoked via `python main.py simulate ...`, argv[1] is 'simulate' -- skip it
raw = sys.argv[1:]
if raw and raw[0] == "simulate":
raw = raw[1:]
args = parser.parse_args(raw)
from db.db import init_db
init_db()
@@ -397,6 +454,7 @@ def main():
spread=args.spread,
slippage=args.slippage,
commission=args.commission,
cap_tier=args.cap_tier,
)
result = simulate(strategy)