Spaces:
Sleeping
Sleeping
| """ | |
| NationStates Card Tracker — Core Module | |
| ======================================== | |
| Shared logic for API calls, database, scanning, and analysis. | |
| Used by both app.py (Gradio UI) and tracker.py (CLI). | |
| """ | |
| import requests | |
| import sqlite3 | |
| import time | |
| import os | |
| import xml.etree.ElementTree as ET | |
| from datetime import datetime | |
| # ========================================== | |
| # CONFIGURATION | |
| # ========================================== | |
| USER_AGENT = "HighValueCardTracker (Operational puppet managed by thunderpuppetempires1)" | |
| API_BASE = "https://www.nationstates.net/cgi-bin/api.cgi" | |
| RATE_LIMIT_SECONDS = 1.2 | |
| SEASON = 4 | |
| MAX_RETRIES = 2 | |
| RETRY_BACKOFF = 2.0 # seconds, doubles each retry | |
| # Price thresholds — alert if ask is below these values | |
| THRESHOLDS = { | |
| "legendary": 20.0, | |
| "epic": 5.0, | |
| "ultra-rare": 1.0, | |
| } | |
| # Historical/famous names in NationStates | |
| HISTORICAL_NAMES = [ | |
| "testlandia", "the pacific", "the north pacific", "the south pacific", | |
| "the east pacific", "the west pacific", "the rejected realms", | |
| "lazarus", "balder", "osiris", "the black hawks", | |
| "the grey wardens", "10000 islands", "europe", | |
| "max barry", "eluvatar", "sedgistan", "violet", | |
| "the free nations region", "the communist bloc", | |
| "nationstates", "maxtopia", "francos spain", "gatesville", | |
| "the pacific army", "jennifer government", "the north pacific army", | |
| "feeder", "raider", "defender", | |
| ] | |
| # ========================================== | |
| # DATABASE | |
| # ========================================== | |
| def init_db(db_path): | |
| """Initialize the SQLite database with all required tables.""" | |
| conn = sqlite3.connect(db_path) | |
| c = conn.cursor() | |
| c.execute(""" | |
| CREATE TABLE IF NOT EXISTS cards ( | |
| card_id INTEGER, | |
| season INTEGER, | |
| name TEXT, | |
| rarity TEXT, | |
| region TEXT, | |
| market_value REAL, | |
| flag TEXT, | |
| PRIMARY KEY (card_id, season) | |
| ) | |
| """) | |
| c.execute(""" | |
| CREATE TABLE IF NOT EXISTS targets ( | |
| card_id INTEGER, | |
| season INTEGER, | |
| name TEXT, | |
| rarity TEXT, | |
| flag_reason TEXT, | |
| last_checked TEXT, | |
| lowest_ask REAL, | |
| card_url TEXT, | |
| PRIMARY KEY (card_id, season) | |
| ) | |
| """) | |
| c.execute(""" | |
| CREATE TABLE IF NOT EXISTS price_history ( | |
| card_id INTEGER, | |
| season INTEGER, | |
| timestamp TEXT, | |
| lowest_ask REAL, | |
| highest_bid REAL, | |
| market_value REAL, | |
| num_asks INTEGER DEFAULT 0, | |
| num_bids INTEGER DEFAULT 0 | |
| ) | |
| """) | |
| c.execute(""" | |
| CREATE TABLE IF NOT EXISTS trade_history ( | |
| card_id INTEGER, | |
| season INTEGER, | |
| timestamp TEXT, | |
| price REAL, | |
| buyer TEXT, | |
| seller TEXT | |
| ) | |
| """) | |
| c.execute(""" | |
| CREATE TABLE IF NOT EXISTS watchlist ( | |
| card_id INTEGER, | |
| season INTEGER, | |
| name TEXT, | |
| rarity TEXT, | |
| max_bid REAL, | |
| notes TEXT, | |
| added_at TEXT, | |
| PRIMARY KEY (card_id, season) | |
| ) | |
| """) | |
| c.execute(""" | |
| CREATE INDEX IF NOT EXISTS idx_price_history_card | |
| ON price_history(card_id, season, timestamp) | |
| """) | |
| conn.commit() | |
| conn.close() | |
| # ========================================== | |
| # API FUNCTIONS (with retry) | |
| # ========================================== | |
| def _api_request(url): | |
| """Make a rate-limited API request with retry/backoff.""" | |
| headers = {"User-Agent": USER_AGENT} | |
| last_error = None | |
| for attempt in range(MAX_RETRIES + 1): | |
| try: | |
| resp = requests.get(url, headers=headers, timeout=15) | |
| time.sleep(RATE_LIMIT_SECONDS) | |
| if resp.status_code == 200: | |
| return resp | |
| elif resp.status_code == 429: | |
| # Rate limited — back off longer | |
| wait = RETRY_BACKOFF * (2 ** attempt) | |
| time.sleep(wait) | |
| last_error = f"HTTP 429 (rate limited)" | |
| continue | |
| elif resp.status_code >= 500: | |
| # Server error — retry | |
| wait = RETRY_BACKOFF * (2 ** attempt) | |
| time.sleep(wait) | |
| last_error = f"HTTP {resp.status_code}" | |
| continue | |
| else: | |
| # Client error (404, etc.) — don't retry | |
| return resp | |
| except requests.exceptions.Timeout: | |
| last_error = "Timeout" | |
| time.sleep(RETRY_BACKOFF * (2 ** attempt)) | |
| except requests.exceptions.ConnectionError: | |
| last_error = "Connection error" | |
| time.sleep(RETRY_BACKOFF * (2 ** attempt)) | |
| except Exception as e: | |
| last_error = str(e) | |
| break | |
| return None, last_error | |
| def api_get(url): | |
| """Make an API request. Returns (response, error_string).""" | |
| result = _api_request(url) | |
| if result is None: | |
| return None, "Unknown error" | |
| if isinstance(result, tuple): | |
| return result | |
| return result, None | |
| def fetch_card_info(card_id, season=SEASON): | |
| """Fetch a single card's info from the API.""" | |
| url = f"{API_BASE}?q=card+info;cardid={card_id};season={season}" | |
| resp, err = api_get(url) | |
| if resp and resp.status_code == 200: | |
| try: | |
| root = ET.fromstring(resp.text) | |
| card = root if root.tag == "CARD" else root.find("CARD") | |
| if card is not None: | |
| return { | |
| "card_id": int(card.findtext("CARDID", str(card_id))), | |
| "season": int(card.findtext("SEASON", str(season))), | |
| "name": card.findtext("NAME", "Unknown"), | |
| "rarity": (card.findtext("CATEGORY", "unknown") or "unknown").lower(), | |
| "region": card.findtext("REGION", ""), | |
| "market_value": float(card.findtext("MARKET_VALUE", "0") or "0"), | |
| "flag": card.findtext("FLAG", ""), | |
| } | |
| except ET.ParseError: | |
| pass | |
| return None | |
| def fetch_card_market(card_id, season=SEASON): | |
| """Fetch the lowest ask price for a card.""" | |
| url = f"{API_BASE}?q=card+markets;cardid={card_id};season={season}" | |
| resp, err = api_get(url) | |
| if resp and resp.status_code == 200: | |
| try: | |
| root = ET.fromstring(resp.text) | |
| markets = root.find(".//MARKETS") | |
| if markets is None: | |
| return None | |
| lowest_ask = None | |
| for market in markets.findall("MARKET"): | |
| if market.findtext("TYPE", "") == "ask": | |
| price = float(market.findtext("PRICE", "0") or "0") | |
| if lowest_ask is None or price < lowest_ask: | |
| lowest_ask = price | |
| return lowest_ask | |
| except (ET.ParseError, ValueError): | |
| pass | |
| return None | |
| def fetch_full_market(card_id, season=SEASON): | |
| """Fetch full market data (all asks and bids) for a card.""" | |
| url = f"{API_BASE}?q=card+markets;cardid={card_id};season={season}" | |
| resp, err = api_get(url) | |
| if resp and resp.status_code == 200: | |
| try: | |
| root = ET.fromstring(resp.text) | |
| markets = root.find(".//MARKETS") | |
| if markets is None: | |
| return {"asks": [], "bids": [], "error": None} | |
| asks = [] | |
| bids = [] | |
| for market in markets.findall("MARKET"): | |
| entry = { | |
| "nation": market.findtext("NATION", ""), | |
| "price": float(market.findtext("PRICE", "0") or "0"), | |
| "timestamp": market.findtext("TIMESTAMP", ""), | |
| } | |
| if market.findtext("TYPE", "") == "ask": | |
| asks.append(entry) | |
| else: | |
| bids.append(entry) | |
| asks.sort(key=lambda x: x["price"]) | |
| bids.sort(key=lambda x: x["price"], reverse=True) | |
| return {"asks": asks, "bids": bids, "error": None} | |
| except (ET.ParseError, ValueError): | |
| pass | |
| return {"asks": [], "bids": [], "error": err or "Failed to fetch market data"} | |
| def fetch_card_trades(card_id, season=SEASON): | |
| """Fetch recent trade history for a card.""" | |
| url = f"{API_BASE}?q=card+trades;cardid={card_id};season={season};limit=20" | |
| resp, err = api_get(url) | |
| if resp and resp.status_code == 200: | |
| try: | |
| root = ET.fromstring(resp.text) | |
| trades_el = root.find(".//TRADES") | |
| if trades_el is None: | |
| return [] | |
| trades = [] | |
| for trade in trades_el.findall("TRADE"): | |
| trades.append({ | |
| "timestamp": trade.findtext("TIMESTAMP", ""), | |
| "price": float(trade.findtext("PRICE", "0") or "0"), | |
| "buyer": trade.findtext("BUYER", ""), | |
| "seller": trade.findtext("SELLER", ""), | |
| }) | |
| return trades | |
| except (ET.ParseError, ValueError): | |
| pass | |
| return [] | |
| def check_nation_exists(nation_name): | |
| """Check if a nation exists via the API.""" | |
| url = f"{API_BASE}?nation={nation_name.replace(' ', '_')}&q=name" | |
| resp, err = api_get(url) | |
| if resp: | |
| return resp.status_code == 200 and "<NAME>" in resp.text | |
| return False | |
| # ========================================== | |
| # SCANNING FUNCTIONS | |
| # ========================================== | |
| def scan_low_id_cards(db_path, max_id=10000, callback=None): | |
| """Scan cards 1 to max_id for Legendary/Epic cards.""" | |
| conn = sqlite3.connect(db_path) | |
| c = conn.cursor() | |
| found = [] | |
| for card_id in range(1, max_id + 1): | |
| card = fetch_card_info(card_id, SEASON) | |
| if card and card["rarity"] in ("legendary", "epic"): | |
| c.execute("INSERT OR REPLACE INTO cards VALUES (?, ?, ?, ?, ?, ?, ?)", | |
| (card["card_id"], card["season"], card["name"], card["rarity"], | |
| card["region"], card["market_value"], card.get("flag", ""))) | |
| card_url = f"https://www.nationstates.net/page=deck/card={card_id}/season={SEASON}" | |
| c.execute("INSERT OR REPLACE INTO targets VALUES (?, ?, ?, ?, ?, ?, ?, ?)", | |
| (card["card_id"], card["season"], card["name"], card["rarity"], | |
| "Low ID (<=10000)", None, None, card_url)) | |
| conn.commit() | |
| found.append(card) | |
| if callback: | |
| callback("found", card_id, card) | |
| else: | |
| if callback: | |
| callback("skip", card_id, card) | |
| conn.close() | |
| return found | |
| def scan_cte_cards(db_path, callback=None): | |
| """Check existing cards for CTE (dead nation) status.""" | |
| conn = sqlite3.connect(db_path) | |
| c = conn.cursor() | |
| c.execute("SELECT card_id, season, name, rarity FROM cards WHERE rarity IN ('legendary', 'epic')") | |
| cards = c.fetchall() | |
| cte_found = [] | |
| for card_id, season, name, rarity in cards: | |
| exists = check_nation_exists(name) | |
| if not exists: | |
| card_url = f"https://www.nationstates.net/page=deck/card={card_id}/season={season}" | |
| c.execute("INSERT OR REPLACE INTO targets VALUES (?, ?, ?, ?, ?, ?, ?, ?)", | |
| (card_id, season, name, rarity, "CTE (Dead Nation)", None, None, card_url)) | |
| conn.commit() | |
| cte_found.append({"card_id": card_id, "name": name, "rarity": rarity}) | |
| if callback: | |
| callback("cte", card_id, name) | |
| else: | |
| if callback: | |
| callback("alive", card_id, name) | |
| conn.close() | |
| return cte_found | |
| def scan_historical_cards(db_path): | |
| """Check cards in DB for historical name matches.""" | |
| conn = sqlite3.connect(db_path) | |
| c = conn.cursor() | |
| c.execute("SELECT card_id, season, name, rarity FROM cards") | |
| cards = c.fetchall() | |
| hist_found = [] | |
| for card_id, season, name, rarity in cards: | |
| for hist_name in HISTORICAL_NAMES: | |
| if hist_name in name.lower(): | |
| card_url = f"https://www.nationstates.net/page=deck/card={card_id}/season={season}" | |
| c.execute("INSERT OR REPLACE INTO targets VALUES (?, ?, ?, ?, ?, ?, ?, ?)", | |
| (card_id, season, name, rarity, | |
| f"Historical ({hist_name})", None, None, card_url)) | |
| conn.commit() | |
| hist_found.append({"card_id": card_id, "name": name, "rarity": rarity, "match": hist_name}) | |
| break | |
| conn.close() | |
| return hist_found | |
| def check_target_prices(db_path, max_checks=50): | |
| """Check live market prices for flagged targets. Returns (alerts, errors).""" | |
| conn = sqlite3.connect(db_path) | |
| c = conn.cursor() | |
| c.execute(""" | |
| SELECT card_id, season, name, rarity, flag_reason FROM targets | |
| ORDER BY CASE rarity WHEN 'legendary' THEN 1 WHEN 'epic' THEN 2 ELSE 3 END | |
| LIMIT ? | |
| """, (max_checks,)) | |
| targets = c.fetchall() | |
| alerts = [] | |
| errors = 0 | |
| for card_id, season, name, rarity, flag_reason in targets: | |
| lowest_ask = fetch_card_market(card_id, season) | |
| now = datetime.now().isoformat() | |
| if lowest_ask is None: | |
| errors += 1 | |
| c.execute("UPDATE targets SET lowest_ask = ?, last_checked = ? WHERE card_id = ? AND season = ?", | |
| (lowest_ask, now, card_id, season)) | |
| conn.commit() | |
| if lowest_ask is not None and lowest_ask > 0: | |
| threshold = THRESHOLDS.get(rarity, 999) | |
| if lowest_ask < threshold: | |
| alerts.append({ | |
| "name": name, "card_id": card_id, "rarity": rarity, | |
| "reason": flag_reason, "ask": lowest_ask, "threshold": threshold, | |
| "url": f"https://www.nationstates.net/page=deck/card={card_id}/season={season}" | |
| }) | |
| conn.close() | |
| return alerts, errors | |
| # ========================================== | |
| # SMART TRADING FUNCTIONS | |
| # ========================================== | |
| def record_price_snapshot(db_path, card_id, season=SEASON): | |
| """Record current market state to price_history.""" | |
| market = fetch_full_market(card_id, season) | |
| card_info = fetch_card_info(card_id, season) | |
| lowest_ask = market["asks"][0]["price"] if market["asks"] else None | |
| highest_bid = market["bids"][0]["price"] if market["bids"] else None | |
| mv = card_info["market_value"] if card_info else None | |
| conn = sqlite3.connect(db_path) | |
| c = conn.cursor() | |
| c.execute("""INSERT INTO price_history (card_id, season, timestamp, lowest_ask, highest_bid, market_value, num_asks, num_bids) | |
| VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", | |
| (card_id, season, datetime.now().isoformat(), lowest_ask, highest_bid, mv, | |
| len(market["asks"]), len(market["bids"]))) | |
| conn.commit() | |
| conn.close() | |
| return { | |
| "lowest_ask": lowest_ask, "highest_bid": highest_bid, "market_value": mv, | |
| "num_asks": len(market["asks"]), "num_bids": len(market["bids"]), | |
| "error": market.get("error") | |
| } | |
| def calculate_trade_score(db_path, card_id, season=SEASON): | |
| """ | |
| Calculate a smart trading score (0-100) for a card. | |
| Higher = better buy opportunity. | |
| """ | |
| conn = sqlite3.connect(db_path) | |
| c = conn.cursor() | |
| c.execute("SELECT rarity, market_value FROM cards WHERE card_id = ? AND season = ?", (card_id, season)) | |
| card_row = c.fetchone() | |
| if not card_row: | |
| conn.close() | |
| return None | |
| rarity, market_value = card_row | |
| c.execute("""SELECT lowest_ask, highest_bid, market_value FROM price_history | |
| WHERE card_id = ? AND season = ? ORDER BY timestamp DESC LIMIT 1""", | |
| (card_id, season)) | |
| latest = c.fetchone() | |
| c.execute("""SELECT lowest_ask FROM price_history | |
| WHERE card_id = ? AND season = ? AND lowest_ask IS NOT NULL | |
| ORDER BY timestamp DESC LIMIT 5""", | |
| (card_id, season)) | |
| price_trend = [r[0] for r in c.fetchall()] | |
| c.execute("SELECT flag_reason FROM targets WHERE card_id = ? AND season = ?", (card_id, season)) | |
| target_row = c.fetchone() | |
| conn.close() | |
| score = 50 # Base score | |
| # Factor 1: Price vs Market Value (0-25 points) | |
| if latest and latest[0] and market_value and market_value > 0: | |
| ask = latest[0] | |
| ratio = ask / market_value | |
| if ratio < 0.5: | |
| score += 25 | |
| elif ratio < 0.8: | |
| score += 15 | |
| elif ratio < 1.0: | |
| score += 8 | |
| elif ratio > 2.0: | |
| score -= 15 | |
| # Factor 2: Price trend (0-15 points) | |
| if len(price_trend) >= 3: | |
| if price_trend[0] < price_trend[-1]: | |
| drop_pct = (price_trend[-1] - price_trend[0]) / price_trend[-1] | |
| score += min(15, int(drop_pct * 50)) | |
| elif price_trend[0] > price_trend[-1]: | |
| score -= 5 | |
| # Factor 3: Spread opportunity (0-10 points) | |
| if latest and latest[0] and latest[1]: | |
| spread = latest[0] - latest[1] | |
| spread_pct = spread / latest[0] if latest[0] > 0 else 0 | |
| if spread_pct > 0.5: | |
| score += 10 | |
| elif spread_pct > 0.3: | |
| score += 6 | |
| elif spread_pct > 0.1: | |
| score += 3 | |
| # Factor 4: Rarity multiplier | |
| rarity_bonus = {"legendary": 15, "epic": 8, "ultra-rare": 4, "rare": 1} | |
| score += rarity_bonus.get(rarity, 0) | |
| # Factor 5: CTE/Historical bonus | |
| if target_row: | |
| reason = target_row[0] | |
| if "CTE" in reason: | |
| score += 10 | |
| if "Historical" in reason: | |
| score += 5 | |
| return max(0, min(100, score)) | |
| def suggest_bid_price(db_path, card_id, season=SEASON): | |
| """Suggest an optimal bid price based on trade history and market state.""" | |
| conn = sqlite3.connect(db_path) | |
| c = conn.cursor() | |
| c.execute("""SELECT price FROM trade_history | |
| WHERE card_id = ? AND season = ? AND price > 0 | |
| ORDER BY timestamp DESC LIMIT 10""", | |
| (card_id, season)) | |
| trade_prices = [r[0] for r in c.fetchall()] | |
| c.execute("""SELECT lowest_ask, highest_bid, market_value FROM price_history | |
| WHERE card_id = ? AND season = ? | |
| ORDER BY timestamp DESC LIMIT 1""", | |
| (card_id, season)) | |
| latest = c.fetchone() | |
| conn.close() | |
| suggestions = {} | |
| if trade_prices: | |
| trade_prices.sort() | |
| median_trade = trade_prices[len(trade_prices) // 2] | |
| suggestions["median_recent_trade"] = round(median_trade, 2) | |
| suggestions["below_median_10pct"] = round(median_trade * 0.9, 2) | |
| if latest: | |
| ask, bid, mv = latest | |
| if ask and bid: | |
| midpoint = (ask + bid) / 2 | |
| suggestions["spread_midpoint"] = round(midpoint, 2) | |
| suggestions["aggressive_bid"] = round(bid + (ask - bid) * 0.3, 2) | |
| suggestions["conservative_bid"] = round(bid + (ask - bid) * 0.1, 2) | |
| if mv and mv > 0: | |
| suggestions["below_mv_20pct"] = round(mv * 0.8, 2) | |
| return suggestions | |