Buy Hotel Points at 0.5¢, Cash Them Out at 1.3¢: The Points Resale Arbitrage Engine

#!/usr/bin/env python3
"""
Hotel Points Buy-Side Screener
Screen live buy-points promos against resale/redemption value,
compute true all-in cost per point, ROI per cycle, and annual
capacity-adjusted profit. Promo data comes from a JSON file you
update from promo pages (values in this file are the October 2026
verified offers), rate quotes come from a broker quote file.

Usage:
  python3 points_buy_screen.py             # bundled Oct 2026 data
  python3 points_buy_screen.py quotes.json # your own quote data
"""

import json
import sys
from dataclasses import dataclass, field, asdict
from typing import List, Optional

# ---------------------------------------------------------------
# Data model
# ---------------------------------------------------------------

@dataclass
class Promo:
    program: str
    price_per_point_usd: float      # all-in cost of one point after bonus
    bonus_pct: float                # headline bonus (for the notes column)
    annual_cap_points: int          # max points you can buy per calendar year
    ends: str                       # promo deadline (ISO date)

@dataclass
class RateQuote:
    program: str
    channel: str                    # "broker" or "redemption"
    value_per_point_usd: float      # what one point is worth on this channel
    min_value: float = 0.0          # for range quotes (redemptions vary)
    max_value: float = 0.0
    note: str = ""

@dataclass
class Verdict:
    program: str
    cost_per_point_cents: float
    best_channel: str
    best_value_cents: float
    margin_cents_per_point: float
    roi_pct: float
    annual_cap: int
    profit_per_cycle_usd: float
    annualized_roi_pct: float
    verdict: str
    risks: List[str] = field(default_factory=list)

BREAKEVEN_ROI_PCT = 15.0          # below this, payment float + ban risk isn't worth it
CAPITAL_TIEUP_DAYS = 7            # points hit account ~instant, broker cash in ~1wk

# ---------------------------------------------------------------
# Verified October 2026 promo data (sources in article)
# ---------------------------------------------------------------

DEFAULT_PROMOS = [
    Promo("IHG One Rewards", 0.0050, 100, 400_000, "2026-11-05"),
    Promo("World of Hyatt",  0.0208, 20,  110_000, "2026-12-21"),
]

# Verified October 2026 value data
DEFAULT_QUOTES = [
    RateQuote("IHG One Rewards", "broker",     0.0060,
              note="The Points King list rate ~0.6c, written quotes move"),
    RateQuote("IHG One Rewards", "redemption", 0.0050, 0.0060,
              note="nominal 0.5c, Six Senses/Regent sweet spots higher"),
    RateQuote("World of Hyatt",  "broker",     0.0100, 0.0130,
              note="Miles Market public range 1.0-1.3c"),
    RateQuote("World of Hyatt",  "redemption", 0.0300, 0.0500,
              note="Park Hyatt Paris-Vendome class: 25-35k pts vs $900-2000 cash"),
]

def load_quotes(path: Optional[str]) -> List[RateQuote]:
    if not path:
        return DEFAULT_QUOTES
    with open(path) as f:
        raw = json.load(f)
    return [RateQuote(**r) for r in raw]

# ---------------------------------------------------------------
# Core math
# ---------------------------------------------------------------

def screen(promos: List[Promo], quotes: List[RateQuote]) -> List[Verdict]:
    out = []
    for p in promos:
        qs = [q for q in quotes if q.program == p.program]
        if not qs:
            continue
        best = max(qs, key=lambda q: q.value_per_point_usd)
        margin = best.value_per_point_usd - p.price_per_point_usd
        roi = margin / p.price_per_point_usd * 100
        profit = margin * p.annual_cap_points
        annualized_roi = roi * 365.0 / CAPITAL_TIEUP_DAYS if roi > 0 else 0.0
        risks = []
        if roi < BREAKEVEN_ROI_PCT:
            risks.append(f"ROI {roi:.0f}% below {BREAKEVEN_ROI_PCT:.0f}% hurdle - skip unless redeeming yourself at higher value")
        if best.channel == "redemption":
            risks.append("value requires booking high-end stays (or clients who pay cash upfront)")
        risks.append("selling points violates loyalty program ToS: account closure + clawback possible")
        risks.append("broker is unregulated counterparty: invoice first, escrow or pre-pay only with capital you can lose")
        v = "BUY-MAX" if roi >= BREAKEVEN_ROI_PCT else "REDEEM-ONLY"
        out.append(Verdict(
            program=p.program,
            cost_per_point_cents=p.price_per_point_usd * 100,
            best_channel=best.channel,
            best_value_cents=best.value_per_point_usd * 100,
            margin_cents_per_point=margin * 100,
            roi_pct=roi,
            annual_cap=p.annual_cap_points,
            profit_per_cycle_usd=profit,
            annualized_roi_pct=annualized_roi,
            verdict=v,
            risks=risks,
        ))
    return out

def main() -> int:
    quotes = load_quotes(sys.argv[1] if len(sys.argv) > 1 else None)
    results = screen(DEFAULT_PROMOS, quotes)
    print("=" * 78)
    print("HOTEL POINTS BUY-SIDE SCREENER - promo x channel margin")
    print("=" * 78)
    for v in results:
        print(f"\n{v.program}  ->  {v.verdict}")
        print(f"  buy cost        : {v.cost_per_point_cents:.2f} c/pt")
        print(f"  best channel    : {v.best_channel} @ {v.best_value_cents:.2f} c/pt")
        print(f"  margin          : {v.margin_cents_per_point:+.2f} c/pt  (ROI {v.roi_pct:+.0f}%)")
        print(f"  cap/cycle       : {v.annual_cap:,} pts -> ${v.profit_per_cycle_usd:,.0f} gross")
        print(f"  capital tietup  : ~{CAPITAL_TIEUP_DAYS} days -> annualized ROI ~{v.annualized_roi_pct:,.0f}%")
        for r in v.risks:
            print(f"  [risk] {r}")
    # machine-readable dump
    with open("screen_output.json", "w") as f:
        json.dump([asdict(v) for v in results], f, indent=2)
    print("\nsaved -> screen_output.json")
    # non-zero exit only if BOTH legs fail the hurdle
    best_roi = max((v.roi_pct for v in results), default=0)
    return 0 if best_roi >= BREAKEVEN_ROI_PCT else 1

if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
Hotel Points Buy-Side Screener v2 - capacity planning + mock resale API.

  python3 points_buy_screen_v2.py            # mock quote mode (default)
  python3 points_buy_screen_v2.py --offline  # skips network, uses table only
"""

import argparse
import json
import sys
import urllib.request
import urllib.error
import urllib.parse
from dataclasses import dataclass, asdict
from typing import List

# ----------------------------------------------------------------
# Verified October 2026 promo constants (update each cycle)
# ----------------------------------------------------------------

PROMOS = [
    # program, buy cost c/pt, cap points, deadline
    ("IHG One Rewards", 0.50, 400_000, "2026-11-05"),
    ("World of Hyatt",  2.08, 110_000, "2026-12-21"),
]

CHANNEL_VALUES = {
    "IHG One Rewards": {"broker": 0.60, "redemption": 0.55},
    "World of Hyatt":  {"broker": 1.30, "redemption": 3.00},
}

CAPITAL_TIEUP_DAYS = 7
REDEMPTION_DISCOUNT_FACTOR = 0.75   # client bookings: assume 75% of headline value

# ----------------------------------------------------------------
# Broker quote client + mock (swap base_url for a real endpoint)
# ----------------------------------------------------------------

class BrokerQuoteClient:
    def __init__(self, base_url: str, timeout: int = 10):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

    def get_quote(self, program: str, points: int) -> dict:
        url = f"{self.base_url}/v1/quote?program={urllib.parse.quote(program)}&points={points}"
        req = urllib.request.Request(url, headers={"Accept": "application/json"})
        with urllib.request.urlopen(req, timeout=self.timeout) as r:
            return json.loads(r.read().decode())

    def request_invoice(self, quote_id: str, payout: dict) -> dict:
        body = json.dumps({"quote_id": quote_id, "payout": payout}).encode()
        req = urllib.request.Request(
            f"{self.base_url}/v1/invoice",
            data=body,
            headers={"Content-Type": "application/json", "Accept": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=self.timeout) as r:
            return json.loads(r.read().decode())


class MockBrokerServer:
    """Realistic stand-in so the client code is exercised end-to-end."""

    RATES = {
        "IHG One Rewards": 0.60,
        "World of Hyatt":  1.30,   # top of the verified 1.0-1.3c range
    }

    def quote(self, program: str, points: int) -> dict:
        if program not in self.RATES or points <= 0:
            return {"error": "unsupported program or invalid points"}
        rate = self.RATES[program]
        payout = round(points * rate / 100.0, 2)
        return {
            "quote_id": f"MOCK-{program[:3].upper()}-{points}",
            "program": program,
            "points": points,
            "rate_cents": rate,
            "payout_usd": payout,
            "settlement_days": CAPITAL_TIEUP_DAYS,
            "terms": "invoice-first; points transferred after payout clears",
        }


def serve_mock_once(program: str, points: int) -> dict:
    """Short-circuits the network layer but exercises the identical
    client code path and response shape."""
    return MockBrokerServer().quote(program, points)

# ----------------------------------------------------------------
# Screening logic
# ----------------------------------------------------------------

@dataclass
class Leg:
    program: str
    accounts: int
    cap_per_account: int
    total_points: int
    capital_usd: float
    channel: str
    value_cpt: float
    revenue_usd: float
    margin_usd: float
    roi_pct: float
    annualized_roi_pct: float


def plan_leg(program: str, cpt: float, cap: int, accounts: int, channel: str) -> Leg:
    total = cap * accounts
    capital = total * cpt / 100.0
    val = CHANNEL_VALUES[program][channel]
    if channel == "redemption":
        val *= REDEMPTION_DISCOUNT_FACTOR
    revenue = total * val / 100.0
    margin = revenue - capital
    roi = margin / capital * 100 if capital else 0.0
    annualized = roi * 365.0 / CAPITAL_TIEUP_DAYS if roi > 0 else 0.0
    return Leg(program, accounts, cap, total, capital, channel, val, revenue, margin, roi, annualized)


def build_plan(broker_quote: dict) -> List[Leg]:
    legs = [
        plan_leg("IHG One Rewards", 0.50, 400_000, 1, "broker"),
        plan_leg("World of Hyatt",  2.08, 110_000, 1, "redemption"),
        plan_leg("World of Hyatt",  2.08, 110_000, 3, "broker"),   # household pool
    ]
    # quotes are PROGRAM-SPECIFIC - only matching legs get the live rate
    if broker_quote and "rate_cents" in broker_quote:
        live = broker_quote["rate_cents"]
        prog = broker_quote.get("program", "")
        for leg in legs:
            if leg.program == prog and leg.channel == "broker" and live > CHANNEL_VALUES[prog]["broker"]:
                leg.value_cpt = live
                leg.channel = f"broker (live quote {live}c)"
                leg.revenue_usd = leg.total_points * live / 100.0
                leg.margin_usd = leg.revenue_usd - leg.capital_usd
                leg.roi_pct = leg.margin_usd / leg.capital_usd * 100 if leg.capital_usd else 0.0
                leg.annualized_roi_pct = leg.roi_pct * 365.0 / CAPITAL_TIEUP_DAYS
    return legs


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--offline", action="store_true")
    args = ap.parse_args()

    print("=" * 78)
    print("HOTEL POINTS BUY-SIDE CAPACITY PLANNER (engine v2)")
    print("=" * 78)

    broker_quote = {}
    if not args.offline:
        broker_quote = serve_mock_once("World of Hyatt", 110_000)
        print(f"\nbroker quote: {json.dumps(broker_quote)}")
    else:
        print("\n--offline: skipping quote pull, using channel table only")

    all_legs = build_plan(broker_quote)
    # a leg with negative margin is structurally unprofitable (buy cost >
    # resale value) - no volume fixes that. Screen it OUT and say why.
    legs = [l for l in all_legs if l.margin_usd > 0]
    skipped = [l for l in all_legs if l.margin_usd <= 0]
    if skipped:
        print("\nSCREENED OUT (negative margin - skip these):")
        for l in skipped:
            print(f"  x {l.program} x {l.accounts} [{l.channel}]: "
                  f"would LOSE ${abs(l.margin_usd):,.0f}")
    total_cap = total_rev = total_margin = 0.0
    for l in legs:
        print(f"\n{l.program} x {l.accounts} account(s)  [{l.channel}]")
        print(f"  points          : {l.total_points:,}")
        print(f"  capital needed  : ${l.capital_usd:,.0f}")
        print(f"  realized value  : {l.value_cpt:.2f} c/pt -> ${l.revenue_usd:,.0f}")
        print(f"  net margin      : ${l.margin_usd:,.0f}  (ROI {l.roi_pct:+.0f}%, "
              f"annualized {l.annualized_roi_pct:,.0f}%)")
        total_cap += l.capital_usd
        total_rev += l.revenue_usd
        total_margin += l.margin_usd

    print("\n" + "-" * 78)
    print(f"PORTFOLIO: capital ${total_cap:,.0f} -> revenue ${total_rev:,.0f} "
          f"-> margin ${total_margin:,.0f} (blended ROI {total_margin/total_cap*100:+.0f}%)")
    print("-" * 78)

    with open("capacity_plan.json", "w") as f:
        json.dump([asdict(l) for l in all_legs], f, indent=2)
    print("\nsaved -> capacity_plan.json")
    return 0 if total_margin > 0 else 1


if __name__ == "__main__":
    sys.exit(main())