Version 1.1

This commit is contained in:
2026-02-18 08:24:54 -05:00
parent 4318c8f642
commit fdba869a8d
33 changed files with 2142 additions and 1942 deletions

573
app.py
View File

@@ -1,172 +1,443 @@
"""
Flask Backend for Lottery Investment Calculator
Provides API endpoints for jackpots and investment calculations
Flask Backend for Lottery Investment Calculator.
API endpoints for jackpots, investment calculations, comparisons,
break-even analysis, annuity projections, and state tax information.
"""
from __future__ import annotations
import logging
from flask import Flask, jsonify, request
from flask_cors import CORS
import requests
from bs4 import BeautifulSoup
import urllib3
from playwright.sync_api import sync_playwright
import re
from lottery_calculator import calculate_us_lottery, calculate_canadian_lottery
# Suppress SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
from config import (
ANNUITY_ANNUAL_INCREASE,
ANNUITY_YEARS,
LOTTERY_ODDS,
STATE_TAX_RATES,
load_config,
)
from lottery_calculator import (
calculate_annuity,
calculate_break_even,
calculate_canadian_lottery,
calculate_group_split,
calculate_us_lottery,
)
from scrapers import clear_cache, get_all_jackpots
app = Flask(__name__)
CORS(app) # Enable CORS for Next.js frontend
# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)
# Common headers to mimic a browser request
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "none",
"Cache-Control": "max-age=0",
}
# ---------------------------------------------------------------------------
# App factory
# ---------------------------------------------------------------------------
def get_us_lotteries():
"""Fetch Powerball and Mega Millions jackpots"""
results = {"Powerball": None, "Mega Millions": None}
# Powerball
try:
resp = requests.get("https://www.lotto.net/powerball", timeout=10, verify=False, headers=HEADERS)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "html.parser")
all_text = soup.get_text()
lines = all_text.split('\n')
for i, line in enumerate(lines):
if 'Next Jackpot' in line and i + 1 < len(lines):
next_line = lines[i + 1].strip()
if '$' in next_line and 'Million' in next_line:
# Extract numeric value
match = re.search(r'\$(\d+(?:,\d+)?(?:\.\d+)?)', next_line)
if match:
value = float(match.group(1).replace(',', ''))
results["Powerball"] = value * 1_000_000
break
except Exception as e:
print(f"Error fetching Powerball: {e}")
# Mega Millions
try:
resp = requests.get("https://www.lotto.net/mega-millions", timeout=10, verify=False, headers=HEADERS)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "html.parser")
all_text = soup.get_text()
lines = all_text.split('\n')
for i, line in enumerate(lines):
if 'Next Jackpot' in line and i + 1 < len(lines):
next_line = lines[i + 1].strip()
if '$' in next_line and 'Million' in next_line:
# Extract numeric value
match = re.search(r'\$(\d+(?:,\d+)?(?:\.\d+)?)', next_line)
if match:
value = float(match.group(1).replace(',', ''))
results["Mega Millions"] = value * 1_000_000
break
except Exception as e:
print(f"Error fetching Mega Millions: {e}")
return results
def create_app() -> Flask:
"""Application factory — creates and configures the Flask app."""
cfg = load_config()
app = Flask(__name__)
def get_canadian_lotteries():
"""Fetch Lotto Max and Lotto 6/49 jackpots using Playwright"""
results = {"Lotto Max": None, "Lotto 6/49": None}
try:
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto("https://www.olg.ca/", wait_until="networkidle", timeout=30000)
page.wait_for_timeout(3000)
content = page.content()
browser.close()
# Lotto Max
lotto_max_match = re.search(r'LOTTO\s*MAX(?:(?!LOTTO\s*6/49).)*?\$\s*([\d.,]+)\s*Million', content, re.IGNORECASE | re.DOTALL)
if lotto_max_match:
value = float(lotto_max_match.group(1).replace(',', ''))
results["Lotto Max"] = value * 1_000_000
# Lotto 6/49
lotto_649_match = re.search(r'LOTTO\s*6/49(?:(?!LOTTO\s*MAX).)*?\$\s*([\d.,]+)\s*Million', content, re.IGNORECASE | re.DOTALL)
if lotto_649_match:
value = float(lotto_649_match.group(1).replace(',', ''))
results["Lotto 6/49"] = value * 1_000_000
except Exception as e:
print(f"Error fetching Canadian lotteries: {e}")
return results
# CORS — restrict origins via env or allow all in dev
if cfg.allowed_origins == "*":
CORS(app)
else:
CORS(app, origins=[o.strip() for o in cfg.allowed_origins.split(",")])
# ------------------------------------------------------------------
# Validation helpers
# ------------------------------------------------------------------
@app.route('/api/jackpots', methods=['GET'])
def get_jackpots():
"""API endpoint to get all lottery jackpots"""
us_lotteries = get_us_lotteries()
canadian_lotteries = get_canadian_lotteries()
return jsonify({
"us": {
"powerball": us_lotteries["Powerball"],
"megaMillions": us_lotteries["Mega Millions"]
},
"canadian": {
"lottoMax": canadian_lotteries["Lotto Max"],
"lotto649": canadian_lotteries["Lotto 6/49"]
def _require_json() -> dict | None:
"""Parse JSON body or return None."""
return request.get_json(silent=True)
def _validate_number(
value, name: str, *, minimum: float = 0, maximum: float | None = None
) -> float | None:
"""Coerce *value* to float and validate range. Returns None on bad input."""
try:
v = float(value)
except (TypeError, ValueError):
return None
if v < minimum:
return None
if maximum is not None and v > maximum:
return None
return v
# ------------------------------------------------------------------
# Jackpot endpoints
# ------------------------------------------------------------------
@app.route("/api/jackpots", methods=["GET"])
def get_jackpots():
"""Return current jackpots for all four lotteries (cached)."""
force = request.args.get("refresh", "").lower() in ("1", "true")
data = get_all_jackpots(force_refresh=force)
return jsonify(data)
@app.route("/api/jackpots/refresh", methods=["POST"])
def refresh_jackpots():
"""Force-refresh the jackpot cache and return new values."""
clear_cache()
data = get_all_jackpots(force_refresh=True)
return jsonify(data)
# ------------------------------------------------------------------
# Calculator endpoints
# ------------------------------------------------------------------
@app.route("/api/calculate", methods=["POST"])
def calculate():
"""Calculate investment returns for a given jackpot."""
data = _require_json()
if data is None:
return jsonify({"error": "Request body must be JSON"}), 400
jackpot = _validate_number(data.get("jackpot"), "jackpot", minimum=1)
if jackpot is None:
return jsonify({"error": "jackpot must be a positive number"}), 400
lottery_type = data.get("type", "us")
if lottery_type not in ("us", "canadian"):
return jsonify({"error": "type must be 'us' or 'canadian'"}), 400
invest_pct = _validate_number(
data.get("investPercentage", cfg.investment.invest_percentage),
"investPercentage",
minimum=0,
maximum=1,
)
annual_return = _validate_number(
data.get("annualReturn", cfg.investment.annual_return),
"annualReturn",
minimum=0,
maximum=1,
)
cycles = _validate_number(
data.get("cycles", cfg.investment.cycles),
"cycles",
minimum=1,
maximum=100,
)
# State tax for US calculations
state_code = data.get("state")
state_tax = cfg.tax.default_state_tax_rate
if lottery_type == "us" and state_code:
state_info = STATE_TAX_RATES.get(state_code.upper())
if state_info:
state_tax = state_info["rate"]
if invest_pct is None or annual_return is None or cycles is None:
return jsonify({"error": "Invalid parameter values"}), 400
try:
if lottery_type == "us":
result = calculate_us_lottery(
jackpot,
invest_percentage=invest_pct,
annual_return=annual_return,
cycles=int(cycles),
state_tax_rate=state_tax,
lump_sum_rate=cfg.tax.lump_sum_rate,
federal_tax_rate=cfg.tax.federal_tax_rate,
usd_cad_rate=cfg.tax.usd_cad_rate,
investment_income_tax_rate=cfg.tax.investment_income_tax_rate,
personal_withdrawal_pct=cfg.tax.personal_withdrawal_pct,
)
else:
result = calculate_canadian_lottery(
jackpot,
invest_percentage=invest_pct,
annual_return=annual_return,
cycles=int(cycles),
investment_income_tax_rate=cfg.tax.investment_income_tax_rate,
personal_withdrawal_pct=cfg.tax.personal_withdrawal_pct,
)
return jsonify(result)
except Exception:
logger.exception("Calculation error")
return jsonify({"error": "Internal calculation error"}), 500
# ------------------------------------------------------------------
# State tax endpoints
# ------------------------------------------------------------------
@app.route("/api/states", methods=["GET"])
def get_states():
"""Return all US states with their lottery tax rates."""
states = [
{"code": code, "name": info["name"], "rate": info["rate"]}
for code, info in sorted(STATE_TAX_RATES.items())
]
return jsonify(states)
@app.route("/api/states/<code>", methods=["GET"])
def get_state(code: str):
"""Return tax info for a specific state."""
info = STATE_TAX_RATES.get(code.upper())
if not info:
return jsonify({"error": f"Unknown state code: {code}"}), 404
return jsonify({"code": code.upper(), "name": info["name"], "rate": info["rate"]})
# ------------------------------------------------------------------
# Comparison endpoint
# ------------------------------------------------------------------
@app.route("/api/compare", methods=["GET"])
def compare():
"""Side-by-side comparison of all lotteries with current jackpots."""
jackpots = get_all_jackpots()
state_code = request.args.get("state")
state_tax = cfg.tax.default_state_tax_rate
if state_code:
st = STATE_TAX_RATES.get(state_code.upper())
if st:
state_tax = st["rate"]
comparisons = []
lottery_map = {
"powerball": ("us", jackpots["us"].get("powerball")),
"megaMillions": ("us", jackpots["us"].get("megaMillions")),
"lottoMax": ("canadian", jackpots["canadian"].get("lottoMax")),
"lotto649": ("canadian", jackpots["canadian"].get("lotto649")),
}
})
for key, (country_type, amount) in lottery_map.items():
odds_info = LOTTERY_ODDS.get(key, {})
entry = {
"key": key,
"name": odds_info.get("name", key),
"country": country_type,
"jackpot": amount,
"odds": odds_info.get("odds"),
"ticketCost": odds_info.get("ticket_cost"),
"oddsFormatted": f"1 in {odds_info.get('odds', 0):,}",
"calculation": None,
}
@app.route('/api/calculate', methods=['POST'])
def calculate():
"""API endpoint to calculate investment returns"""
data = request.json
jackpot = data.get('jackpot')
lottery_type = data.get('type', 'us') # 'us' or 'canadian'
invest_percentage = data.get('investPercentage', 0.90)
annual_return = data.get('annualReturn', 0.045)
cycles = data.get('cycles', 8)
if not jackpot:
return jsonify({"error": "Jackpot amount is required"}), 400
try:
if lottery_type == 'us':
result = calculate_us_lottery(jackpot, invest_percentage, annual_return, cycles)
else:
result = calculate_canadian_lottery(jackpot, invest_percentage, annual_return, cycles)
if amount and amount > 0:
try:
if country_type == "us":
calc = calculate_us_lottery(
amount,
state_tax_rate=state_tax,
lump_sum_rate=cfg.tax.lump_sum_rate,
federal_tax_rate=cfg.tax.federal_tax_rate,
usd_cad_rate=cfg.tax.usd_cad_rate,
investment_income_tax_rate=cfg.tax.investment_income_tax_rate,
personal_withdrawal_pct=cfg.tax.personal_withdrawal_pct,
)
else:
calc = calculate_canadian_lottery(
amount,
investment_income_tax_rate=cfg.tax.investment_income_tax_rate,
personal_withdrawal_pct=cfg.tax.personal_withdrawal_pct,
)
entry["calculation"] = calc
except Exception:
logger.exception("Comparison calc failed for %s", key)
comparisons.append(entry)
return jsonify(comparisons)
# ------------------------------------------------------------------
# Break-even calculator
# ------------------------------------------------------------------
@app.route("/api/calculate/breakeven", methods=["POST"])
def break_even():
"""Calculate the jackpot amount where expected value equals ticket cost."""
data = _require_json()
if data is None:
return jsonify({"error": "Request body must be JSON"}), 400
lottery_key = data.get("lottery", "powerball")
odds_info = LOTTERY_ODDS.get(lottery_key)
if not odds_info:
return jsonify({"error": f"Unknown lottery: {lottery_key}"}), 400
ticket_cost = _validate_number(
data.get("ticketCost", odds_info["ticket_cost"]),
"ticketCost",
minimum=0.01,
)
state_code = data.get("state")
state_tax = cfg.tax.default_state_tax_rate
if state_code:
st = STATE_TAX_RATES.get(state_code.upper())
if st:
state_tax = st["rate"]
result = calculate_break_even(
odds=odds_info["odds"],
ticket_cost=ticket_cost,
country=odds_info["country"],
lump_sum_rate=cfg.tax.lump_sum_rate,
federal_tax_rate=cfg.tax.federal_tax_rate,
state_tax_rate=state_tax,
)
result["lottery"] = odds_info["name"]
result["oddsFormatted"] = f"1 in {odds_info['odds']:,}"
return jsonify(result)
except Exception as e:
return jsonify({"error": str(e)}), 500
# ------------------------------------------------------------------
# Annuity calculator
# ------------------------------------------------------------------
@app.route("/api/calculate/annuity", methods=["POST"])
def annuity():
"""Calculate 30-year annuity payout schedule."""
data = _require_json()
if data is None:
return jsonify({"error": "Request body must be JSON"}), 400
jackpot = _validate_number(data.get("jackpot"), "jackpot", minimum=1)
if jackpot is None:
return jsonify({"error": "jackpot must be a positive number"}), 400
lottery_type = data.get("type", "us")
state_code = data.get("state")
state_tax = cfg.tax.default_state_tax_rate
if state_code:
st = STATE_TAX_RATES.get(state_code.upper())
if st:
state_tax = st["rate"]
years = int(
_validate_number(
data.get("years", ANNUITY_YEARS), "years", minimum=1, maximum=40
)
or ANNUITY_YEARS
)
annual_increase = (
_validate_number(
data.get("annualIncrease", ANNUITY_ANNUAL_INCREASE),
"annualIncrease",
minimum=0,
maximum=0.20,
)
or ANNUITY_ANNUAL_INCREASE
)
result = calculate_annuity(
jackpot=jackpot,
country=lottery_type,
years=years,
annual_increase=annual_increase,
federal_tax_rate=cfg.tax.federal_tax_rate,
state_tax_rate=state_tax,
)
return jsonify(result)
# ------------------------------------------------------------------
# Group play calculator
# ------------------------------------------------------------------
@app.route("/api/calculate/group", methods=["POST"])
def group_play():
"""Split winnings among a group with optional custom shares."""
data = _require_json()
if data is None:
return jsonify({"error": "Request body must be JSON"}), 400
jackpot = _validate_number(data.get("jackpot"), "jackpot", minimum=1)
if jackpot is None:
return jsonify({"error": "jackpot must be a positive number"}), 400
members_val = _validate_number(
data.get("members", 2), "members", minimum=1, maximum=1000
)
members = int(members_val) if members_val else 2
shares = data.get("shares") # optional list of floats summing to 1.0
lottery_type = data.get("type", "us")
state_code = data.get("state")
state_tax = cfg.tax.default_state_tax_rate
if state_code:
st = STATE_TAX_RATES.get(state_code.upper())
if st:
state_tax = st["rate"]
result = calculate_group_split(
jackpot=jackpot,
members=members,
shares=shares,
country=lottery_type,
lump_sum_rate=cfg.tax.lump_sum_rate,
federal_tax_rate=cfg.tax.federal_tax_rate,
state_tax_rate=state_tax,
usd_cad_rate=cfg.tax.usd_cad_rate,
investment_income_tax_rate=cfg.tax.investment_income_tax_rate,
personal_withdrawal_pct=cfg.tax.personal_withdrawal_pct,
)
return jsonify(result)
# ------------------------------------------------------------------
# Odds / probability info
# ------------------------------------------------------------------
@app.route("/api/odds", methods=["GET"])
def get_odds():
"""Return odds and ticket cost for all supported lotteries."""
result = []
for key, info in LOTTERY_ODDS.items():
result.append(
{
"key": key,
"name": info["name"],
"odds": info["odds"],
"oddsFormatted": f"1 in {info['odds']:,}",
"oddsPercentage": f"{(1 / info['odds']) * 100:.10f}%",
"ticketCost": info["ticket_cost"],
"country": info["country"],
}
)
return jsonify(result)
# ------------------------------------------------------------------
# Health check
# ------------------------------------------------------------------
@app.route("/api/health", methods=["GET"])
def health():
"""Health check endpoint."""
return jsonify({"status": "ok", "version": "2.0.0"})
return app
@app.route('/api/health', methods=['GET'])
def health():
"""Health check endpoint"""
return jsonify({"status": "ok"})
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
app = create_app()
if __name__ == '__main__':
print("🎰 Lottery Investment Calculator API")
print("=" * 50)
print("Starting Flask server on http://localhost:5000")
print("API Endpoints:")
print(" - GET /api/jackpots - Get current jackpots")
print(" - POST /api/calculate - Calculate investments")
print(" - GET /api/health - Health check")
print("=" * 50)
# Bind to 0.0.0.0 so the Flask app is reachable from outside the container
app.run(debug=True, host='0.0.0.0', port=5000)
if __name__ == "__main__":
cfg = load_config()
logger.info("Lottery Investment Calculator API v2.0")
logger.info("Endpoints:")
logger.info(" GET /api/jackpots - Current jackpots (cached)")
logger.info(" POST /api/jackpots/refresh - Force refresh")
logger.info(" POST /api/calculate - Investment calculator")
logger.info(" POST /api/calculate/breakeven - Break-even calculator")
logger.info(" POST /api/calculate/annuity - Annuity calculator")
logger.info(" POST /api/calculate/group - Group play calculator")
logger.info(" GET /api/compare - Side-by-side comparison")
logger.info(" GET /api/states - US state tax rates")
logger.info(" GET /api/odds - Lottery odds info")
logger.info(" GET /api/health - Health check")
app.run(debug=cfg.debug, host=cfg.host, port=cfg.port)