Gig Quest Stacking: The Python Optimizer That Turns Delivery App Bonuses Into $28/Hour Runs

#!/usr/bin/env python3
"""
Gig Quest Stacking Optimizer
============================
Helps delivery drivers (DoorDash / Uber Eats / Grubhub) decide:
  1. Which app quests/challenges are worth chasing (marginal value math)
  2. Whether a Peak Pay + Challenge stack clears your $/hour floor
  3. Which live offers to accept vs auto-decline ($/min + $/mile grade)
  4. Which 3-hour blocks to schedule this week based on your own history

IMPORTANT: The driver history in this file is MOCK data for demo purposes.
Point load_history() at your real CSV export (columns: date, hour, deliveries,
dollars) to run it on your own numbers. No API keys required.

Run:  python3 quest_stack_optimizer.py
"""

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional
import random
import sys

# --------------------------------------------------------------------------
# Tunables - change these to your market
# --------------------------------------------------------------------------
TARGET_USD_PER_HOUR = 25.00     # walk-away floor; decline work below this
GAS_PER_MILE = 0.76             # 2026 IRS standard mileage rate (Jul 1-Dec 31)
                                # used as the true per-mile cost proxy
MAX_DEADHEAD_MILES = 3.0        # extra miles you'll drive for one order


# --------------------------------------------------------------------------
# 1. QUEST EVALUATION - marginal value beats headline value
# --------------------------------------------------------------------------
@dataclass
class Quest:
    platform: str
    name: str
    deliveries_required: int
    bonus_usd: float
    window_days: int
    deadline_hours_away: float


def marginal_bonus_per_extra_delivery(
    quest: Quest, baseline_deliveries_in_window: int
) -> Optional[float]:
    """$ per EXTRA delivery the quest actually forces you to do.

    If your natural volume already covers most of the quest, the payout is
    spread over far fewer extra runs - that's when quests are free money.
    Returns None if the quest is not feasible at your baseline volume.
    """
    extra_needed = quest.deliveries_required - baseline_deliveries_in_window
    if extra_needed <= 0:
        return float("inf")  # already satisfied: pure payout, zero extra work
    deliveries_per_day = baseline_deliveries_in_window / max(quest.window_days, 1)
    days_available = quest.deadline_hours_away / 24.0
    feasible_capacity = deliveries_per_day * days_available * 1.35  # 35% push
    if extra_needed > feasible_capacity:
        return None  # not realistically completable -> decline enrollment
    return quest.bonus_usd / extra_needed


def quest_verdict(mv: Optional[float], quest: Quest) -> str:
    if mv is None:
        return "SKIP - volume gap too big for the window"
    if mv == float("inf"):
        return "TAKE - zero extra deliveries needed"
    if mv >= 6.0:
        return "TAKE - strong marginal value"
    if mv >= 3.5:
        return "TAKE if you already dash that window anyway"
    return "SKIP - bonus spread too thin"


def evaluate_quests(quests: List[Quest], history: List[dict]) -> List[dict]:
    results = []
    for q in quests:
        baseline = baseline_volume(q.window_days, history)
        mv = marginal_bonus_per_extra_delivery(q, baseline)
        results.append({
            "platform": q.platform,
            "quest": q.name,
            "needs": q.deliveries_required,
            "bonus_usd": q.bonus_usd,
            "baseline_volume": baseline,
            "marginal_usd_per_extra_delivery": (
                "FREE (already covered)" if mv == float("inf")
                else "INFEASIBLE - skip" if mv is None
                else round(mv, 2)
            ),
            "verdict": quest_verdict(mv, q),
        })
    return results


# --------------------------------------------------------------------------
# 2. STACK MATH - blended $/hour with Peak Pay + Challenge amortized in
# --------------------------------------------------------------------------
def blended_hourly(
    base_per_delivery: float,
    tip_per_delivery: float,
    peak_pay_per_delivery: float,
    deliveries_per_hour: float,
    active_challenge: Optional[Quest] = None,
) -> Dict[str, float]:
    """Challenge bonus is amortized over the deliveries you'll do inside
    its window while also collecting Peak Pay on the same completions
    (DoorDash confirms a single delivery counts toward both)."""
    per_delivery = base_per_delivery + tip_per_delivery + peak_pay_per_delivery
    challenge_amortized = 0.0
    if active_challenge:
        in_window = deliveries_per_hour * active_challenge.deadline_hours_away
        extra = max(0, min(in_window, active_challenge.deliveries_required))
        if extra:
            challenge_amortized = (
                active_challenge.bonus_usd / extra
            )
        # cap amortization at the quest's flat per-delivery value
        challenge_amortized = min(
            challenge_amortized,
            active_challenge.bonus_usd / active_challenge.deliveries_required,
        )
    gross = (per_delivery + challenge_amortized) * deliveries_per_hour
    return {
        "per_delivery_cash": round(per_delivery, 2),
        "challenge_amortized_per_delivery": round(challenge_amortized, 2),
        "gross_per_hour": round(gross, 2),
    }


# --------------------------------------------------------------------------
# 3. OFFER GRADER - accept/decline in 3 seconds
# --------------------------------------------------------------------------
def grade_offer(pay_usd: float, miles: float, est_minutes: float) -> Dict[str, object]:
    """Rule: keep offers >= target $/hour AND >= $1.50/mile."""
    cost = miles * GAS_PER_MILE
    net = pay_usd - cost
    usd_per_min = net / est_minutes if est_minutes > 0 else 0.0
    usd_per_hour_equiv = usd_per_min * 60
    usd_per_mile = pay_usd / miles if miles > 0 else 999
    accept = (
        usd_per_hour_equiv >= TARGET_USD_PER_HOUR
        and usd_per_mile >= 1.50
    )
    return {
        "net_usd": round(net, 2),
        "usd_per_mile": round(usd_per_mile, 2),
        "hourly_equivalent": round(usd_per_hour_equiv, 2),
        "decision": "ACCEPT" if accept else "DECLINE",
    }


# --------------------------------------------------------------------------
# 4. BLOCK PLANNER - your best hours, from your own history
# --------------------------------------------------------------------------
def baseline_volume(window_days: int, history: List[dict]) -> int:
    cutoff = datetime.now(timezone.utc) - timedelta(days=window_days)
    total = 0
    for row in history:
        try:
            d = datetime.strptime(row["date"], "%Y-%m-%d").replace(
                tzinfo=timezone.utc
            )
        except (KeyError, ValueError):
            continue
        if d >= cutoff:
            total += int(row.get("deliveries", 0))
    return total


def best_blocks(history: List[dict], block_hours: int = 3, top_n: int = 3) -> List[dict]:
    """Aggregate earnings by starting hour across history, return best windows."""
    hourly: Dict[int, Dict[str, float]] = {}
    for row in history:
        h = int(row.get("hour", -1))
        if h < 0:
            continue
        slot = hourly.setdefault(h, {"dollars": 0.0, "deliveries": 0.0, "samples": 0})
        slot["dollars"] += float(row.get("dollars", 0))
        slot["deliveries"] += float(row.get("deliveries", 0))
        slot["samples"] += 1

    scored = []
    for start, agg in hourly.items():
        window_dollars = 0.0
        window_hours = 0
        for offset in range(block_hours):
            slot = hourly.get((start + offset) % 24)
            if slot and slot["samples"] > 0:
                window_dollars += slot["dollars"] / slot["samples"]
                window_hours += 1
        if window_hours < block_hours:
            continue  # incomplete history for this window
        scored.append({
            "block": f"{start:02d}:00-{(start + block_hours) % 24:02d}:00",
            "avg_usd_per_hour": round(window_dollars / block_hours, 2),
        })
    scored.sort(key=lambda x: x["avg_usd_per_hour"], reverse=True)
    return scored[:top_n]


# --------------------------------------------------------------------------
# MOCK DATA (demo) - replace with your real export
# --------------------------------------------------------------------------
def load_history() -> List[dict]:
    """Demo history: 6 weeks of dash logs. In production, replace the body of
    this function with csv.DictReader(open('my_dash_log.csv')) where columns
    are: date,hour,deliveries,dollars"""
    random.seed(42)
    history = []
    # lunch peak 11-13, dinner peak 17-20, dead 14-16
    pattern = {10: 16, 11: 27, 12: 25, 13: 19, 16: 18, 17: 29, 18: 31, 19: 26, 20: 21}
    start = (datetime.now(timezone.utc) - timedelta(days=42)).replace(
        tzinfo=timezone.utc
    )
    for day in range(42):
        d = start + timedelta(days=day)
        if d.weekday() >= 5 and random.random() < 0.4:
            continue  # drivers skip some weekends
        for hour, avg_usd in pattern.items():
            if random.random() < 0.15:
                continue  # sometimes you skip a slot
            usd = avg_usd * random.uniform(0.75, 1.3)
            deliveries = round(usd / random.uniform(6.5, 9.0))
            history.append({
                "date": d.strftime("%Y-%m-%d"),
                "hour": hour,
                "deliveries": deliveries,
                "dollars": round(usd, 2),
            })
    return history


def main() -> int:
    history = load_history()
    # For quest evaluation use a "lazy week" baseline (what you actually did
    # last week, roughly half of your peak capacity) - typical for most drivers
    eval_history = [
        dict(r, deliveries=max(1, int(r["deliveries"] * 0.5)))
        for r in history if r["hour"] in (17, 18, 19, 20)
    ]

    print("=" * 70)
    print("GIG QUEST STACKING OPTIMIZER - demo run with MOCK history")
    print(f"Target floor: ${TARGET_USD_PER_HOUR:.2f}/hr | "
          f"cost basis: ${GAS_PER_MILE:.2f}/mi (IRS 2026)")
    print("=" * 70)

    # --- 1. Quest board (read these from your Opportunities / Promos tab) ---
    quests = [
        Quest("DoorDash", "Challenge: 15 deliveries by Sunday", 15, 20.0, 7, 96),
        Quest("DoorDash", "Challenge: 40 deliveries by Sunday", 40, 45.0, 7, 96),
        Quest("Uber Eats", "Quest menu: 25 trips / $35", 25, 35.0, 5, 72),
        Quest("Uber Eats", "Quest menu: 60 trips / $110", 60, 110.0, 5, 72),
        Quest("Grubhub", "Peak block bonus: 12 orders tonight", 12, 18.0, 1, 10),
    ]
    print("\n[1] QUEST EVALUATION (marginal value = bonus / EXTRA deliveries)")
    print("-" * 70)
    for r in evaluate_quests(quests, eval_history):
        print(f"  {r['platform']:<10} | {r['quest']:<38} | "
              f"{str(r['marginal_usd_per_extra_delivery']):>24} | {r['verdict']}")

    # --- 2. Stack math for tonight's dinner run ---
    print("\n[2] STACKED HOUR - dinner window, Peak Pay $4 + $1 combined, "
          "active $20/15 challenge")
    print("-" * 70)
    stack = blended_hourly(
        base_per_delivery=3.50,          # inside DD's official $2-10+ base band
        tip_per_delivery=6.00,
        peak_pay_per_delivery=5.00,      # two stacked incentives ($4 + $1)
        deliveries_per_hour=1.8,
        active_challenge=quests[0],      # $20 / 15 deliveries challenge
    )
    for k, v in stack.items():
        print(f"  {k:<36} {v}")
    verdict = ("GO - above floor" if stack["gross_per_hour"] >= TARGET_USD_PER_HOUR
               else "MARGINAL - at floor")
    print(f"  {'VERDICT':<36} {verdict}")

    # --- 3. Live offer grading ---
    print("\n[3] OFFER GRADER (3 offers in queue)")
    print("-" * 70)
    offers = [
        ("Offer A", 7.25, 2.1, 24),
        ("Offer B", 12.50, 4.8, 42),
        ("Offer C", 13.25, 1.4, 26),
    ]
    for name, pay, miles, mins in offers:
        g = grade_offer(pay, miles, mins)
        print(f"  {name}: ${pay:.2f} / {miles:.1f}mi / {mins}min -> "
              f"net ${g['net_usd']}, ${g['usd_per_mile']}/mi, "
              f"${g['hourly_equivalent']}/hr equiv -> {g['decision']}")

    # --- 4. Best blocks from your own logged history ---
    print("\n[4] BEST 3-HOUR BLOCKS (from 6-week mock history)")
    print("-" * 70)
    for b in best_blocks(history):
        print(f"  {b['block']}  avg ${b['avg_usd_per_hour']}/hr")

    print("\nDone - all four modules executed cleanly.")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as exc:  # fail loudly, never silently
        print(f"FATAL: {exc}", file=sys.stderr)
        sys.exit(1)

import csv
from datetime import datetime, timezone

def load_history():
    history = []
    with open("my_dash_log.csv", newline="") as f:
        for row in csv.DictReader(f):
            history.append({
                "date": datetime.strptime(row["date"], "%Y-%m-%d")
                        .replace(tzinfo=timezone.utc).strftime("%Y-%m-%d"),
                "hour": int(row["hour"]),
                "deliveries": int(row["deliveries"]),
                "dollars": float(row["dollars"]),
            })
    return history