dodoxandClaude Sonnet 5
c4178fe779
Finds manufactured goods currently selling below their implied production cost (live input cost vs output price) - candidates to buy and hold for a price-reversion bounce, the mirror of the "is this worth building" question saturation.py already answers. Backtested first (326 output tickers, 287 deduplicated dip-onset episodes across 112 distinct tickers): +3mo forward return mean +21.3%/median +8.9%/73.2% hit rate at >=10 units/day volume - real signal, not noise, but only ~0.10-0.24%/day, far below the 14-29%/day marginal ROI this session's production builds found. Documented prominently in the tool's own docstring and CLAUDE.md so it doesn't get mistaken for a primary strategy - it's for excess idle capital once better-yielding builds are exhausted, not before. Promoted daily_prices() out of tools/history.py into puga/history.py so both tools share it, per the self-improve rule. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
71 lines
3.4 KiB
Python
71 lines
3.4 KiB
Python
"""Finds manufactured goods currently selling below their implied production cost (output price
|
|
minus input cost, from live daily VWAP candles) - the mirror image of saturation.py's "is this
|
|
worth building" question: here we're asking "is this worth buying and holding for a price-
|
|
reversion bounce."
|
|
|
|
Backtested 2026-10-09 across 326 output tickers / 112 distinct tickers that ever showed the
|
|
signal, 287 independent dip-onset episodes (deduplicated so a persisting dip only counts once,
|
|
at onset, not once per month it continues): at >=10 units/day traded volume, +3 months forward
|
|
price return was mean +21.3%, median +8.9%, 73.2% of episodes positive (see empire/docs or the
|
|
session that built this for the raw numbers). That is ~0.10-0.24%/day - MUCH lower than this
|
|
project's production-building marginal ROI (14-29%/day seen this session). Only worth deploying
|
|
*excess* capital into once better-yielding builds are exhausted, not as a primary strategy, and
|
|
it's a gross-return backtest with no spread/freight/storage-fee/capital-lockup cost subtracted.
|
|
"""
|
|
from . import fio
|
|
from .history import daily_prices
|
|
|
|
|
|
def manufacturing_candidates():
|
|
"""One representative recipe per output ticker, excluding pure-extraction buildings (no
|
|
material inputs to compute an implied cost from)."""
|
|
seen = {}
|
|
for r in fio.recipes():
|
|
if r["BuildingTicker"] in ("EXT", "COL", "RIG") or not r["Inputs"]:
|
|
continue
|
|
for o in r["Outputs"]:
|
|
seen.setdefault(o["Ticker"], r)
|
|
return seen
|
|
|
|
|
|
def current_margin(ticker: str, recipe: dict, cx: str, window_days: int = 35, recent_days: int = 14):
|
|
"""Mean margin%/price/volume over the most recent `recent_days` of overlapping price history
|
|
for this recipe's inputs+output, within a `window_days` lookback. None if not enough history."""
|
|
ins = {i["Ticker"]: i["Amount"] for i in recipe["Inputs"]}
|
|
outs = {o["Ticker"]: o["Amount"] for o in recipe["Outputs"]}
|
|
series = {m: daily_prices(m, cx) for m in list(ins) + list(outs)}
|
|
if any(not s for s in series.values()):
|
|
return None
|
|
days = sorted(set().union(*[set(s) for s in series.values()]))[-window_days:]
|
|
last = {}
|
|
rows = []
|
|
for d in days:
|
|
for m, s in series.items():
|
|
if d in s:
|
|
last[m] = s[d][0]
|
|
if len(last) == len(series):
|
|
rev = sum(last[m] * n for m, n in outs.items())
|
|
cost = sum(last[m] * n for m, n in ins.items())
|
|
rows.append((rev, cost, last[ticker], series[ticker].get(d, (0, 0))[1]))
|
|
recent = rows[-recent_days:]
|
|
if len(recent) < recent_days // 2:
|
|
return None
|
|
rev_m = sum(x[0] for x in recent) / len(recent)
|
|
cost_m = sum(x[1] for x in recent) / len(recent)
|
|
px_m = sum(x[2] for x in recent) / len(recent)
|
|
vol_m = sum(x[3] for x in recent) / len(recent)
|
|
margin_pct = 100 * (rev_m - cost_m) / cost_m if cost_m else 0.0
|
|
return {"margin_pct": margin_pct, "price": px_m, "volume": vol_m, "building": recipe["BuildingTicker"]}
|
|
|
|
|
|
def scan_current(cx: str, min_vol: float = 5.0):
|
|
"""Every manufactured good currently priced below its implied production cost, with enough
|
|
volume to actually trade, sorted most-undervalued first."""
|
|
out = []
|
|
for t, r in manufacturing_candidates().items():
|
|
m = current_margin(t, r, cx)
|
|
if m and m["margin_pct"] < 0 and m["volume"] >= min_vol:
|
|
out.append({"ticker": t, **m})
|
|
out.sort(key=lambda x: x["margin_pct"])
|
|
return out
|