"""Core trading engine: the periodic loop that drives every platform.
The engine owns the tick: it polls contracts, matches trade requests, decides
when to open offers, funds escrows and records completed trades.
Note on this public repository
------------------------------
The operative trading logic — offer creation and sizing decisions, escrow
funding coordination, trade-request matching and auto-accept, and premium
maintenance on stale offers — is not published here. Those methods are kept as
documented signatures so the architecture, the control flow and the module
boundaries stay readable, but they raise NotImplementedError.
Everything else in this repository is complete and real, including the taproot
escrow implementation, BIP32 key derivation, offer sizing arithmetic, the
platform and exchange clients, and the data model. The intent is to show how
the system is built without shipping a bot that runs unattended against a live
marketplace and real money.
"""
import json, logging, os, threading, time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
import requests as _requests
from core.models import OfferStatus
from core.fileutil import atomic_write_json
from core.trade_logger import TradeLogger
from core.pricing import DynamicPricer
log = logging.getLogger("bot.engine")
# Transient errors worth retrying (network issues, server overload)
_RETRYABLE = (ConnectionError, TimeoutError, OSError,
_requests.ConnectionError, _requests.Timeout, _requests.HTTPError)
def _is_retryable(exc):
"""Check if an exception is transient (worth retrying) vs fatal (config/logic)."""
if isinstance(exc, _RETRYABLE):
# 4xx (except 429) are NOT retryable — bad request, auth, not found
if isinstance(exc, _requests.HTTPError) and exc.response is not None:
code = exc.response.status_code
return code >= 500 or code == 429
return True
# String check fallback for wrapped exceptions
s = str(exc).lower()
return any(kw in s for kw in ("timeout", "connection", "503", "502", "429", "retry"))
def _mempool_get(url, timeout=10, retries=3):
"""GET from mempool.space with retry + blockstream.info fallback."""
import requests as _req
last_err = None
for attempt in range(retries):
try:
r = _req.get(url, timeout=timeout)
if r.status_code >= 500 and attempt < retries - 1:
time.sleep(2 ** attempt); continue
r.raise_for_status()
return r.json()
except (_req.ConnectionError, _req.Timeout) as e:
last_err = e
if attempt < retries - 1: time.sleep(2 ** attempt)
# Fallback to blockstream.info
fallback_url = url.replace("https://mempool.space/api", "https://blockstream.info/api")
if fallback_url != url:
try:
r = _req.get(fallback_url, timeout=timeout)
r.raise_for_status()
return r.json()
except Exception:
pass
raise last_err or Exception("mempool.space + blockstream.info request failed")
class SpotPriceProvider:
_cache = {} # {currency: (price, timestamp)} — fallback if all sources fail
@staticmethod
def _cached(currency, price):
SpotPriceProvider._cache[currency] = (price, time.time())
return price
@staticmethod
def _get_cached(currency, max_age=120):
"""Return cached price if fresh enough (default 2min), else None.
Hard limit: reject cache older than 5min to prevent stale-price trades."""
HARD_MAX_AGE = 300 # 5 minutes absolute maximum
entry = SpotPriceProvider._cache.get(currency)
if not entry:
return None
age = time.time() - entry[1]
if age > HARD_MAX_AGE:
log.error(f"Cached {currency} spot price too old ({age:.0f}s), rejecting")
return None
if age < max_age:
log.warning(f"Using cached {currency} spot price ({age:.0f}s old)")
return entry[0]
return None
@staticmethod
def get_spot_eur():
import requests
try:
r = requests.get("https://api.coingecko.com/api/v3/simple/price", params={"ids":"bitcoin","vs_currencies":"eur"}, timeout=10)
r.raise_for_status()
return SpotPriceProvider._cached("EUR", r.json()["bitcoin"]["eur"])
except Exception as e:
log.debug(f"CoinGecko EUR: {e}")
try:
r = requests.get("https://api.kraken.com/0/public/Ticker", params={"pair":"XXBTZEUR"}, timeout=10)
r.raise_for_status()
for v in r.json()["result"].values(): return SpotPriceProvider._cached("EUR", float(v["c"][0]))
except Exception as e:
log.debug(f"Kraken EUR: {e}")
cached = SpotPriceProvider._get_cached("EUR")
if cached: return cached
raise ValueError("Cannot fetch spot price")
@staticmethod
def get_spot_usd():
import requests
try:
r = requests.get("https://api.kraken.com/0/public/Ticker", params={"pair":"XBTUSD"}, timeout=10)
r.raise_for_status()
for v in r.json()["result"].values(): return SpotPriceProvider._cached("USD", float(v["c"][0]))
except Exception as e:
log.debug(f"Kraken USD: {e}")
try:
r = requests.get("https://api.coingecko.com/api/v3/simple/price", params={"ids":"bitcoin","vs_currencies":"usd"}, timeout=10)
r.raise_for_status()
return SpotPriceProvider._cached("USD", r.json()["bitcoin"]["usd"])
except Exception as e:
log.debug(f"CoinGecko USD: {e}")
cached = SpotPriceProvider._get_cached("USD")
if cached: return cached
raise ValueError("Cannot fetch USD spot price")
@staticmethod
def get_spot_chf():
import requests
try:
r = requests.get("https://api.coingecko.com/api/v3/simple/price", params={"ids":"bitcoin","vs_currencies":"chf"}, timeout=10)
r.raise_for_status()
return SpotPriceProvider._cached("CHF", r.json()["bitcoin"]["chf"])
except Exception as e:
log.debug(f"CoinGecko CHF: {e}")
try:
eur = SpotPriceProvider.get_spot_eur()
r = requests.get("https://api.kraken.com/0/public/Ticker", params={"pair":"EURCHF"}, timeout=10)
r.raise_for_status()
for v in r.json()["result"].values(): return SpotPriceProvider._cached("CHF", eur * float(v["c"][0]))
except Exception as e:
log.debug(f"Kraken EURCHF: {e}")
cached = SpotPriceProvider._get_cached("CHF")
if cached: return cached
raise ValueError("Cannot fetch CHF spot price")
@staticmethod
def get_spot(currency="EUR"):
if currency == "EUR": return SpotPriceProvider.get_spot_eur()
elif currency == "USD": return SpotPriceProvider.get_spot_usd()
elif currency == "CHF": return SpotPriceProvider.get_spot_chf()
else: return SpotPriceProvider.get_spot_eur()
class TradingEngine:
def __init__(self, config):
self.config = config
self._config_path = os.path.join(os.path.dirname(__file__), "..", "config.json")
self._config_lock = threading.Lock() # serializes config.json writes
self.platforms = {}
self.exchanges = {}
self.trade_logger = TradeLogger(config.get("db_path","trades.db"))
self.pricer = DynamicPricer(config)
self.notifier = None
self.pending_escrows = {}
self._escrow_lock = threading.Lock() # protects pending_escrows access
# Share lock with fund_from_wallet to prevent concurrent UTXO spending
try:
from fund_from_wallet import set_funding_lock
set_funding_lock(self._escrow_lock)
except ImportError:
pass
self._contracted_offers = set() # Offer IDs that have active contracts (don't re-sync, max 100)
self._contracted_offers_file = os.path.join(os.path.dirname(__file__), "..", "contracted_offers.json")
self._load_contracted_offers()
self._contracted_offers_snapshot = set(self._contracted_offers)
self.daily_volume_sats = 0
self.paused = False
self.poll_interval = config.get("poll_interval",30)
self._poll_active = config.get("poll_interval_active", 12) # funded offers exist
self._poll_idle = config.get("poll_interval_idle", 60) # no funded offers
self.daily_volume_limit = config.get("daily_volume_limit_usd",2000.0)
self.auto_fund = config.get("auto_buy_and_fund",False)
self.auto_confirm = config.get("auto_confirm_payment",False)
self._notified_contracts = set() # track already-notified contract IDs
self._recorded_contracts = set() # track already-logged trades
self._low_balance_warned = {} # {exchange_name: float (timestamp)} — per-exchange last warning time
self._buy_data_cache = {} # preserved buy data after offer removed from pending_escrows
self._start_time = datetime.now()
self._last_daily_summary = None
self._last_weekly_summary = None
self._last_monthly_summary = None
self._last_yearly_summary = None
self._pending_refunds = {} # {offer_id: {escrow_addr, amount_sats, created}}
self._last_refund_check = None
self._last_stale_check = None
self._last_consolidation_check = None
self._last_auto_buy_check = 0
self._last_offer_created_at = 0
self._auto_buy_amount_index = 0
self._last_auto_fund_wallet_check = 0
self._last_contracted_offers_save = 0
self._platform_down_since = {} # {name: timestamp} — when platform first became unreachable
self._platform_down_notified = set() # platforms where outage alert was already sent
self._escrow_state_file = os.path.join(os.path.dirname(__file__), "..", "escrow_state.json")
self._escrow_state = self._load_escrow_state()
self._cached_active_offers = {} # {platform_name: [offers]} — reused by _auto_broadcast_refunds
def reload_config(self, config_path=None):
"""Hot-reload config.json — updates engine config, exchanges, and platform credentials."""
if config_path is None:
config_path = os.path.join(os.path.dirname(__file__), "..", "config.json")
with open(config_path) as f:
new_cfg = json.load(f)
self.config = new_cfg
# Reinitialize exchanges with new API keys
from exchanges.kraken import KrakenExchange
for key, ecfg in new_cfg.get("exchanges", {}).items():
if not ecfg.get("enabled"): continue
if not key.startswith("kraken"): continue
ecfg.setdefault("name", key)
ex_name = ecfg["name"]
try:
new_ex = KrakenExchange(ecfg)
self.exchanges[ex_name] = new_ex
log.info(f"reload_config: exchange {ex_name} reinitialized")
except Exception as e:
log.warning(f"reload_config: exchange {ex_name}: {e}")
# Update Peach platform credentials without full re-init
peach = self.platforms.get("peach")
if peach:
pcfg = new_cfg.get("platforms", {}).get("peach", {})
for attr in ("payment_data_raw", "payment_data_info", "pgp_private_key",
"pgp_public_key", "refund_address"):
val = pcfg.get(attr)
if val is not None:
setattr(peach, attr, val)
peach.access_token = None # force re-auth on next tick
log.info("reload_config: peach credentials updated, re-auth scheduled")
# Dynamic pricing thresholds are held in the pricer's own config objects, so they
# would otherwise survive a reload unchanged while it still logs success.
pricer = getattr(self, "pricer", None)
if pricer is not None and hasattr(pricer, "reload"):
try:
pricer.reload(new_cfg)
log.info("reload_config: dynamic pricing thresholds updated")
except Exception as e:
log.warning(f"reload_config: pricer reload: {e}")
log.info("Config reloaded successfully")
def save_config(self):
"""Persist the in-memory config to config.json atomically, under a lock.
Single source of truth — callers mutate self.config then call this. Avoids the
in-memory/disk divergence and format drift from ad-hoc writes (e.g. a premium set over
Telegram getting lost on restart), and writing the WHOLE config matters: the /auto
toggles touch both auto_buy_escrow and platforms.peach.sepa_accounts, so saving just one
section would silently drop the other. External edits to config.json must be followed by
reload_config() (e.g. via SIGHUP) so the in-memory copy stays authoritative.
"""
with self._config_lock:
atomic_write_json(self._config_path, self.config, indent=2)
def add_platform(self, p):
self.platforms[p.name] = p
log.info(f"Platform: {p.name}")
def add_exchange(self, e):
self.exchanges[e.name] = e
log.info(f"Exchange: {e.name}")
def set_notifier(self, n): self.notifier = n
def add_pending_escrow(self, offer_id, escrow_addr, amount_sats, premium, sepa_account_index=0):
"""Register a new offer in pending_escrows and log it. Thread-safe."""
with self._escrow_lock:
self.pending_escrows[offer_id] = {
"platform": "peach", "escrow_address": escrow_addr,
"amount_sats": amount_sats, "funded": False,
"funding_in_progress": True, "premium": premium,
"sepa_account_index": sepa_account_index}
# Persist sepa_account_index immediately so it survives restarts
# (premium reduction only saves it after 12h — too late if bot restarts before that)
self._escrow_state[str(offer_id)] = {
"premium": premium,
"sepa_account_index": sepa_account_index,
# Recorded so a re-published offer can be linked back to this one — see
# _adopt_republished_state. The escrow survives a re-publish; the offer id does not.
"escrow_address": escrow_addr or "",
}
self._save_escrow_state()
self._last_offer_created_at = time.time()
self.trade_logger.log_event("peach", "offer_created", offer_id)
def _adopt_republished_state(self, offer_id, old_offer_id=None, escrow_address=None):
"""Carry our local record over when Peach re-publishes an offer under a new id.
Peach can re-publish an offer under a NEW id (the new offer carries `oldOfferId`) while
keeping the same funded escrow. Everything we know about the offer — buy_data, premium,
SEPA account — is keyed by the OLD id, so without this the trade is booked from a
fallback estimate.
Links via `oldOfferId` first, then via the escrow address, which is the one thing a
re-publish cannot change. Fields already recorded under the new id win. Returns the
adopted state, or None when there is nothing to adopt.
"""
oid = str(offer_id)
if self._escrow_state.get(oid, {}).get("buy_data"):
return None # already a complete record under this id
src = None
if old_offer_id and str(old_offer_id) in self._escrow_state:
src = str(old_offer_id)
elif escrow_address:
src = next((k for k, v in self._escrow_state.items()
if k != oid and v.get("escrow_address") == escrow_address), None)
if not src:
return None
state = {**self._escrow_state[src], **self._escrow_state.get(oid, {}), "republished_from": src}
if escrow_address:
state["escrow_address"] = escrow_address
self._escrow_state[oid] = state
self._save_escrow_state()
log.info(f"peach: offer {oid} is a re-publish of {src} — carried over buy data, premium and account")
return state
def save_buy_data(self, offer_id, buy_data: dict):
"""Persist buy_data to escrow_state so it survives bot restarts."""
with self._escrow_lock:
if offer_id in self.pending_escrows:
self.pending_escrows[offer_id]["buy_data"] = buy_data
state = self._escrow_state.setdefault(str(offer_id), {})
state["buy_data"] = buy_data
self._save_escrow_state()
def get_best_exchange(self):
best = None
best_price = float("inf")
for ex in self.exchanges.values():
try:
price = ex.get_spot_price()
if price < best_price:
best_price = price
best = ex
except Exception as e:
log.warning(f"{ex.name}: {e}")
return best
def get_exchange_by_currency(self, currency):
"""Get exchange instance that trades in the given fiat currency."""
for ex in self.exchanges.values():
if hasattr(ex, 'get_fiat_currency') and ex.get_fiat_currency() == currency:
return ex
return None
def start(self):
log.info("="*50)
log.info(f" Platforms: {list(self.platforms.keys())}")
log.info(f" Exchanges: {list(self.exchanges.keys())}")
log.info(f" Auto-fund: {self.auto_fund}")
log.info("="*50)
for name, p in self.platforms.items():
try:
p.authenticate()
log.info(f"{name}: auth OK")
except Exception as e:
log.error(f"{name}: auth failed: {e}")
for name, e in self.exchanges.items():
try:
s = e.get_status()
currency = s.get('currency', 'CHF')
log.info(f"{name}: {s.get('fiat_balance',0):.2f} {currency}")
except Exception as e2:
log.warning(f"{name}: {e2}")
self._load_pending_refunds()
_consecutive_errors = 0
while True:
try:
if not self.paused: self._tick()
_consecutive_errors = 0
except KeyboardInterrupt:
log.info("Shutting down...")
break
except Exception as e:
_consecutive_errors += 1
if _is_retryable(e):
log.warning(f"Tick transient error ({_consecutive_errors}x): {e}")
else:
log.error(f"Tick fatal error: {e}")
if self.notifier: self.notifier.notify_error(str(e))
# Back off on repeated errors: 1x=normal, 2x=double, max 5min
if _consecutive_errors >= 3:
backoff = min(self.poll_interval * _consecutive_errors, 300)
log.warning(f"Backing off {backoff}s after {_consecutive_errors} consecutive errors")
time.sleep(backoff)
continue
# Adaptive polling: faster when funded offers exist (waiting for matches)
with self._escrow_lock:
has_funded = any(v.get("funded") for v in self.pending_escrows.values())
interval = self._poll_active if has_funded else self._poll_idle
time.sleep(interval)
def _prune_tracking_sets(self):
"""Prevent unbounded growth of tracking sets.
The limits must stay above the number of terminal contracts, otherwise this
defeats the very filter it feeds: a contract is skipped only while its id is
in _recorded_contracts, so pruning to 100 meant several hundred finished
contracts were re-read and re-recorded on every single tick. The written
state file still looked correct, because the set grew back within the same
tick. Kept in sync with the caps in _save_id_set.
"""
for s, label, keep in ((self._notified_contracts, "_notified_contracts", 5000),
(self._recorded_contracts, "_recorded_contracts", 20000)):
if len(s) > keep:
# Keep newest (contract IDs are roughly chronological)
sorted_ids = sorted(s)
to_remove = sorted_ids[:-keep]
s -= set(to_remove)
log.debug(f"Pruned {label}: {len(to_remove)} old entries removed")
def _tick(self):
tick_start = time.time()
for name, platform in self.platforms.items():
try: self._process_platform(name, platform)
except Exception as e: log.error(f"{name}: {e}")
self._check_daily_summary()
self._check_weekly_summary()
self._check_monthly_summary()
self._check_yearly_summary()
self._check_low_balance()
self._check_pending_refunds()
self._auto_consolidate_utxos()
self._auto_buy_escrow_check()
self._auto_fund_wallet_check()
self._prune_tracking_sets()
tick_duration = time.time() - tick_start
if tick_duration > self.poll_interval:
log.warning(f"Tick took {tick_duration:.1f}s (> {self.poll_interval}s poll interval)")
def _check_daily_summary(self):
now = datetime.now()
if now.hour == 22 and (self._last_daily_summary is None or self._last_daily_summary.date() < now.date()):
self._last_daily_summary = now
if self.notifier:
try:
since = datetime(now.year, now.month, now.day).isoformat()
s = self.trade_logger.get_since_summary(since, "Tagesrückblick")
b = self.trade_logger.get_method_breakdown(since)
self.notifier.notify_period_summary(s, "Tagesrückblick", "📆", b)
except Exception as e:
log.debug(f"Daily summary: {e}")
def _check_weekly_summary(self):
now = datetime.now()
if now.weekday() != 6 or now.hour != 22: return # Sonntag 22:00
week = now.isocalendar()[1]
if self._last_weekly_summary and self._last_weekly_summary.isocalendar()[1] == week:
return
self._last_weekly_summary = now
if self.notifier:
try:
since = (now - timedelta(days=7)).isoformat()
w = self.trade_logger.get_since_summary(since, "Wochenrückblick")
b = self.trade_logger.get_method_breakdown(since)
self.notifier.notify_period_summary(w, "Wochenrückblick", "📊", b)
except Exception as e:
log.debug(f"Weekly summary: {e}")
def _check_monthly_summary(self):
import calendar
now = datetime.now()
last_day = calendar.monthrange(now.year, now.month)[1]
if now.day != last_day or now.hour != 22: return
month_key = (now.year, now.month)
if self._last_monthly_summary and (self._last_monthly_summary.year, self._last_monthly_summary.month) == month_key:
return
self._last_monthly_summary = now
if self.notifier:
try:
since = datetime(now.year, now.month, 1).isoformat()
month_name = now.strftime("%B %Y")
s = self.trade_logger.get_since_summary(since, month_name)
b = self.trade_logger.get_method_breakdown(since)
self.notifier.notify_period_summary(s, f"Monatsrückblick {month_name}", "📅", b)
except Exception as e:
log.debug(f"Monthly summary: {e}")
def _check_yearly_summary(self):
now = datetime.now()
if now.month != 12 or now.day != 31 or now.hour != 22: return
if self._last_yearly_summary and self._last_yearly_summary.year == now.year:
return
self._last_yearly_summary = now
if self.notifier:
try:
since = datetime(now.year, 1, 1).isoformat()
s = self.trade_logger.get_since_summary(since, str(now.year))
b = self.trade_logger.get_method_breakdown(since)
self.notifier.notify_period_summary(s, f"Jahresrückblick {now.year}", "🎆", b)
except Exception as e:
log.debug(f"Yearly summary: {e}")
def _check_low_balance(self):
now = datetime.now()
if now.minute not in (0, 1): return # check at top of hour (tolerant of 30s tick)
threshold = self.config.get("low_balance_threshold", 50)
cooldown = self.config.get("low_balance_cooldown_sec", 21600) # 6h default
for ex in self.exchanges.values():
try:
fiat = ex.get_fiat_balance(cached=True) # use cache — avoids extra API call every hour
ex_name = ex.name if hasattr(ex, 'name') else str(ex)
last_warned = self._low_balance_warned.get(ex_name, 0)
if fiat < threshold and (not last_warned or time.time() - last_warned > cooldown):
self._low_balance_warned[ex_name] = time.time()
if self.notifier:
currency = ex.get_fiat_currency() if hasattr(ex, 'get_fiat_currency') else 'EUR'
self.notifier.notify_low_balance(fiat, currency)
log.warning(f"Low balance {ex_name}: {fiat:.2f} {currency}")
elif fiat >= threshold:
self._low_balance_warned.pop(ex_name, None) # reset when balance recovers
except Exception:
pass
def _rotation_amounts(self, cfg):
"""The EUR amounts the rotation may use this cycle.
In cap_fraction mode these are derived from Peach's cap at the current
spot, so offers stay as close to the cap as Peach allows instead of
drifting with the BTC price. Peach denominates that cap in CHF, so the
CHF spot is needed too; without it the sizing falls back to the
`max_offer_sats` cap rather than guessing. Falls back to the configured
list if the EUR spot price is unavailable, so a price-feed hiccup cannot
silently shrink every offer.
"""
from core.offer_sizing import effective_amounts
configured = sorted([a for a in cfg.get("amounts", [200, 300, 400, 500, 600])
if isinstance(a, (int, float))])
if cfg.get("size_mode", "fixed") != "cap_fraction":
return configured
try:
spot = SpotPriceProvider.get_spot_eur()
except Exception:
return configured
pconf = self.config.get("platforms", {}).get("peach", {})
fee = cfg.get("withdraw_fee_sats", 2000)
try:
spot_chf = SpotPriceProvider.get_spot_chf()
except Exception:
spot_chf = None # resolve_cap_sats then uses the max_offer_sats fallback
amounts = effective_amounts(cfg, spot, fee, pconf.get("min_amount_sats", 10000), spot_chf)
return amounts or configured
def _auto_buy_escrow_check(self):
"""Decide whether to open a new offer this cycle, and at which size.
Applies the size rotation, the concurrency and spacing limits, and the
available fiat balance, then reserves the sats the resulting offer will draw
before handing off to the buy pipeline.
Not included in the public repository — see the note at the top of this
module.
"""
raise NotImplementedError(
"_auto_buy_escrow_check is not part of the public repository")
def _auto_fund_wallet_check(self):
"""Move confirmed hot-wallet sats into escrows awaiting funding.
Accounts for sats already reserved by other pending offers so two offers can
never be funded from the same UTXOs, and caps the amount at the marketplace
limit.
Not included in the public repository — see the note at the top of this
module.
"""
raise NotImplementedError(
"_auto_fund_wallet_check is not part of the public repository")
def _auto_consolidate_utxos(self):
"""Consolidate hot wallet UTXOs when idle and fees are low. Runs every 30 min."""
now = datetime.now()
if self._last_consolidation_check and (now - self._last_consolidation_check).total_seconds() < 1800:
return
self._last_consolidation_check = now
# Only consolidate when no escrow funding is actively in progress (avoid UTXO conflicts).
# Funded offers waiting for a buyer match don't use UTXOs — safe to consolidate.
with self._escrow_lock:
funding_in_progress = any(v.get("funding_in_progress") for v in self.pending_escrows.values())
if funding_in_progress:
return
try:
from fund_from_wallet import consolidate_utxos
result = consolidate_utxos(self.config)
if result:
log.info(f"UTXO consolidation: merged {result['utxos_merged']} UTXOs, fee {result['fee']} sats, txid {result['txid']}")
if self.notifier:
self.notifier._send(
f"UTXO Konsolidierung\n"
f"{result['utxos_merged']} UTXOs → 1\n"
f"Total: {result['total_sats']:,} sats\n"
f"Fee: {result['fee']:,} sats\n"
f"TXID: {result['txid'][:16]}..."
)
except Exception as e:
log.warning(f"UTXO consolidation: {e}")
def _load_pending_refunds(self):
"""Load pending refunds from file (persists across restarts). Thread-safe."""
import os
refund_file = os.path.join(os.path.dirname(__file__), "..", "pending_refunds.json")
with self._escrow_lock:
if os.path.exists(refund_file):
try:
with open(refund_file) as f:
self._pending_refunds = json.load(f)
if self._pending_refunds:
total = sum(v["amount_sats"] for v in self._pending_refunds.values())
log.info(f"Loaded {len(self._pending_refunds)} pending refunds ({total:,} sats)")
except Exception as e:
log.warning(f"Failed to load pending refunds: {e}")
def register_refund(self, offer_id, platform_name="peach"):
"""Register a cancelled offer for refund monitoring. Fetches escrow info from Peach."""
if str(offer_id) in self._pending_refunds:
return # already tracked
platform = self.platforms.get(platform_name)
if not platform:
return
try:
platform._ensure_auth()
r = platform.session.get(f"{platform.base_url}/offer/{offer_id}/escrow", timeout=15)
if r.status_code != 200:
log.debug(f"No escrow info for {offer_id}: HTTP {r.status_code}")
return
d = r.json()
escrow_addr = d.get("escrow", "")
amounts = d.get("funding", {}).get("amounts", [])
amount = amounts[0] if amounts else 0
if not escrow_addr or amount <= 0:
log.debug(f"Offer {offer_id}: no funded escrow")
return
# Check if still has balance (skip if already refunded)
md = _mempool_get(f"https://mempool.space/api/address/{escrow_addr}")
balance = (md.get("chain_stats", {}).get("funded_txo_sum", 0)
- md.get("chain_stats", {}).get("spent_txo_sum", 0)
+ md.get("mempool_stats", {}).get("funded_txo_sum", 0)
- md.get("mempool_stats", {}).get("spent_txo_sum", 0))
if balance <= 0:
log.info(f"Offer {offer_id}: escrow already empty, no refund tracking needed")
return
self._pending_refunds[str(offer_id)] = {
"escrow_addr": escrow_addr,
"amount_sats": amount,
"created": datetime.now().isoformat()
}
self._save_pending_refunds()
log.info(f"Tracking refund for offer {offer_id}: {amount:,} sats at {escrow_addr[:30]}...")
if self.notifier:
self.notifier._send(
f"Refund-Monitoring\n"
f"Offer {offer_id}: {amount:,} sats\n"
f"Warte auf Rückzahlung..."
)
except Exception as e:
log.warning(f"Failed to register refund for {offer_id}: {e}")
def _save_pending_refunds(self):
"""Save pending refunds to file (thread-safe via _escrow_lock)."""
import os
refund_file = os.path.join(os.path.dirname(__file__), "..", "pending_refunds.json")
with self._escrow_lock:
try:
with open(refund_file, "w") as f:
json.dump(self._pending_refunds, f, indent=2)
except Exception as e:
log.debug(f"Failed to save pending refunds: {e}")
def _check_pending_refunds(self):
if not self._pending_refunds:
self._auto_broadcast_refunds(cached_offers=self._cached_active_offers)
return
now = datetime.now()
# Check every 5 minutes
if self._last_refund_check and (now - self._last_refund_check).total_seconds() < 300:
return
self._last_refund_check = now
completed = []
for oid, info in list(self._pending_refunds.items()):
try:
addr = info["escrow_addr"]
d = _mempool_get(f"https://mempool.space/api/address/{addr}")
funded = d.get("chain_stats", {}).get("funded_txo_sum", 0)
spent = d.get("chain_stats", {}).get("spent_txo_sum", 0)
balance = funded - spent
mfunded = d.get("mempool_stats", {}).get("funded_txo_sum", 0)
mspent = d.get("mempool_stats", {}).get("spent_txo_sum", 0)
balance += mfunded - mspent
if balance <= 0:
log.info(f"Refund received for offer {oid} ({info['amount_sats']} sats)")
completed.append(oid)
if self.notifier:
self.notifier._send(
f"Refund erhalten!\n"
f"Offer {oid}: {info['amount_sats']:,} sats zurück"
)
except Exception as e:
log.debug(f"Refund check {oid}: {e}")
for oid in completed:
del self._pending_refunds[oid]
if completed:
self._save_pending_refunds()
if self._pending_refunds:
remaining = sum(v["amount_sats"] for v in self._pending_refunds.values())
log.debug(f"Pending refunds: {len(self._pending_refunds)} offers, {remaining:,} sats")
if completed and not self._pending_refunds:
log.info("All pending refunds received!")
if self.notifier:
self.notifier._send("Alle Refunds erhalten!")
# Also check for unbroadcast refund PSBTs
self._auto_broadcast_refunds()
def _auto_broadcast_refunds(self, cached_offers=None):
"""Sign and broadcast any unbroadcast refund PSBTs from Peach."""
now = datetime.now()
# Check every 10 minutes
if not hasattr(self, '_last_refund_broadcast_check'):
self._last_refund_broadcast_check = None
if self._last_refund_broadcast_check and (now - self._last_refund_broadcast_check).total_seconds() < 600:
return
self._last_refund_broadcast_check = now
for pname, platform in self.platforms.items():
try:
platform._ensure_auth()
offers = cached_offers.get(pname) if cached_offers else None
if offers is None:
offers = platform.get_active_offers()
for o in offers:
raw = o.raw_data
refund_psbt = raw.get("refundTx", "")
refunded = raw.get("refunded", False)
if not refund_psbt or refunded:
continue
oid = raw.get("id", "")
amount = raw.get("amount", 0)
try:
from coincurve import PrivateKey
from release_escrow import build_finalized_tx
import requests as _req
# Get derivation path from escrow status
esc = platform.get_escrow_status(oid)
deriv = esc.get("funding", {}).get("derivationPath", "")
deriv_id = int(deriv.split("/")[-1]) if deriv else int(oid)
key_hex = platform._get_escrow_privkey_hex(deriv_id)
privkey = PrivateKey(bytes.fromhex(key_hex))
tx_hex = build_finalized_tx(refund_psbt, privkey)
if not tx_hex:
continue
r = _req.post("https://mempool.space/api/tx", data=tx_hex, timeout=15)
if r.status_code == 200:
log.info(f"{pname}: auto-refund broadcast for offer {oid} ({amount:,} sats)")
if self.notifier:
self.notifier._send(
f"Refund gebroadcastet!\n"
f"Offer {oid}: {amount:,} sats\n"
f"TX: {r.text[:16]}..."
)
else:
# Already spent or other error — not critical
log.debug(f"{pname}: refund broadcast {oid}: {r.text[:100]}")
except Exception as e:
log.debug(f"{pname}: auto-refund {oid}: {e}")
except Exception as e:
log.debug(f"{pname}: auto-refund check: {e}")
def _timed(self, label, fn, *args):
"""Run fn and log duration if >1s."""
t0 = time.time()
try:
return fn(*args)
finally:
dt = time.time() - t0
if dt > 1.0:
log.info(f"⏱ {label}: {dt:.1f}s")
else:
log.debug(f"⏱ {label}: {dt:.2f}s")
def _process_platform(self, name, platform):
# Quick health check: if auth fails, skip entire platform this tick
try:
platform._ensure_auth()
# Platform reachable — clear outage state if it was down
if name in self._platform_down_since:
down_sec = int(time.time() - self._platform_down_since.pop(name))
self._platform_down_notified.discard(name)
log.info(f"{name}: API wieder erreichbar (war {down_sec}s down)")
if self.notifier:
self.notifier._send(f"✅ {name} API wieder erreichbar\n(war {down_sec}s nicht verfügbar)")
except Exception as e:
log.warning(f"{name}: auth failed, skipping tick: {e}")
now = time.time()
if name not in self._platform_down_since:
self._platform_down_since[name] = now
down_sec = now - self._platform_down_since[name]
if down_sec > 300 and name not in self._platform_down_notified:
self._platform_down_notified.add(name)
log.error(f"{name}: API seit {down_sec:.0f}s nicht erreichbar!")
if self.notifier:
self.notifier._send(
f"🚨 {name} API nicht erreichbar\n"
f"Seit {int(down_sec // 60)} Minuten ausgefallen.\n"
f"Fehler: {str(e)[:200]}")
return
# Sequential: create offers + fund first (must complete before match checking)
self._timed(f"{name}/offers", self._create_offers, name, platform)
if self.auto_fund: self._timed(f"{name}/fund", self._fund_escrows, name, platform)
# Parallel: contracts, matches, stale checks are independent
with ThreadPoolExecutor(max_workers=3, thread_name_prefix="tick") as pool:
futures = {
pool.submit(self._timed, f"{name}/contracts", self._check_contracts, name, platform): "contracts",
pool.submit(self._timed, f"{name}/matches", self._check_matches, name, platform): "matches",
pool.submit(self._timed, f"{name}/stale", self._check_stale_offers, name, platform): "stale",
}
for f in as_completed(futures):
try:
f.result()
except Exception as e:
log.error(f"{name}/{futures[f]}: {e}")
def _create_offers(self, name, platform):
"""Create the sell offers for a platform and register their escrows.
Not included in the public repository — see the note at the top of this
module.
"""
raise NotImplementedError(
"_create_offers is not part of the public repository")
# Offers are created manually via /buy_escrow — no auto-create limit
def _fund_escrows(self, name, platform):
"""Fund the escrows of offers that are waiting for coins.
Not included in the public repository — see the note at the top of this
module.
"""
raise NotImplementedError(
"_fund_escrows is not part of the public repository")
def _check_matches(self, name, platform):
"""Poll incoming trade requests and auto-accept the ones that qualify.
Resolves the authoritative live-offer list, filters out offers that are
terminal or already contracted, matches each request against the published
payment data, and accepts under the configured constraints.
Not included in the public repository — see the note at the top of this
module.
"""
raise NotImplementedError(
"_check_matches is not part of the public repository")
def _check_contracts(self, name, platform):
# Prune _contracted_offers to prevent unbounded growth (keep newest 50 by lexicographic sort)
if len(self._contracted_offers) > 100:
self._contracted_offers = set(sorted(self._contracted_offers)[-50:])
try: contracts = platform.get_contracts()
except Exception as e: log.warning(f"{name}: contracts: {e}"); return
for c in contracts:
# When a contract exists for an offer, remove offer from pending_escrows to stop trade request polling
# But preserve buy_data for profit calculation
offer_id = (c.id.split('-')[0] if '-' in c.id else getattr(c, 'offer_id', None)) or ''
if offer_id:
self._contracted_offers.add(offer_id)
with self._escrow_lock:
removed = self.pending_escrows.pop(offer_id, None)
if removed and removed.get("buy_data"):
self._buy_data_cache[offer_id] = removed["buy_data"]
# Prevent unbounded cache growth
if len(self._buy_data_cache) > 50:
oldest = list(self._buy_data_cache.keys())[0]
self._buy_data_cache.pop(oldest, None)
if c.status == OfferStatus.PAYMENT_RECEIVED:
if c.id not in self._notified_contracts:
log.info(f"{name}: PAYMENT RECEIVED {c.id[:16]}")
self._notified_contracts.add(c.id)
if self.auto_confirm:
try:
# Try to get PSBT and sign for release
if hasattr(platform, 'sign_release_psbt'):
psbt_data = platform.sign_release_psbt(c.id)
if psbt_data:
log.info(f"{name}: PSBT ready for {c.id[:16]} - manual signing needed")
if self.notifier:
self.notifier._send(f"PSBT Ready!\nContract {c.id[:16]}\nSign and release manually or auto-release will attempt.")
# Simple confirm (without release TX)
platform.confirm_payment(c.id)
if c.id not in self._recorded_contracts:
self._recorded_contracts.add(c.id)
self._record_trade(name, c)
except Exception as e:
log.error(f"{name}: confirm: {e}")
elif c.status == OfferStatus.DISPUTE:
# The de-dup key must be SEPARATE from the payment-received one. Both used the
# bare contract id in `_notified_contracts`, and a dispute always follows a
# payment-received for the same contract — so the guard was already satisfied and
# the dispute message never went out. One contract sat unnoticed for two days
# that way, while the same line was logged 3550 times.
_dkey = f"dispute:{c.id}"
if _dkey in self._notified_contracts:
log.debug(f"{name}: DISPUTE {c.id} (already reported)")
else:
raw = c.raw_data or {}
log.warning(f"{name}: DISPUTE {c.id} — {c.amount_sats} sats, "
f"{c.price_fiat} {c.currency}, {c.payment_method}")
if self.notifier:
_acct = (self._escrow_state.get(str(c.offer_id), {}) or {}).get("sepa_account_name", "")
self.notifier.notify_dispute(c.id, {
"amount_sats": c.amount_sats, "price": c.price_fiat,
"currency": c.currency, "method": c.payment_method,
"account": _acct, "since": raw.get("lastModified", ""),
})
self._notified_contracts.add(_dkey)
elif c.status == OfferStatus.COMPLETED:
with self._escrow_lock:
for oid in (c.offer_id, c.id):
if oid and oid in self.pending_escrows:
removed = self.pending_escrows.pop(oid)
if removed.get("buy_data"):
self._buy_data_cache[oid] = removed["buy_data"]
break
if c.id not in self._recorded_contracts:
self._recorded_contracts.add(c.id)
self._record_trade(name, c)
if self._contracted_offers != self._contracted_offers_snapshot:
self._save_contracted_offers()
self._contracted_offers_snapshot = set(self._contracted_offers)
def _load_contracted_offers(self):
try:
if os.path.exists(self._contracted_offers_file):
with open(self._contracted_offers_file) as f:
self._contracted_offers = set(json.load(f))
log.info(f"Loaded {len(self._contracted_offers)} contracted offers from disk")
except Exception as e:
log.debug(f"Failed to load contracted offers: {e}")
def _save_contracted_offers(self):
"""Write contracted offers to disk — debounced to max once per 60s."""
now = time.time()
if now - self._last_contracted_offers_save < 60:
return
self._last_contracted_offers_save = now
try:
ids = sorted(self._contracted_offers)
if len(ids) > 100:
ids = ids[-100:]
self._contracted_offers = set(ids)
with open(self._contracted_offers_file, "w") as f:
json.dump(ids, f)
except Exception as e:
log.debug(f"Failed to save contracted offers: {e}")
def _load_escrow_state(self):
"""Load persisted escrow state (funded_at timers after premium reductions)."""
try:
if os.path.exists(self._escrow_state_file):
with open(self._escrow_state_file) as f:
return json.load(f)
except Exception as e:
log.debug(f"Failed to load escrow state: {e}")
return {}
def _save_escrow_state(self):
"""Save escrow state so premium reduction timers survive restarts."""
try:
with open(self._escrow_state_file, "w") as f:
json.dump(self._escrow_state, f, indent=2)
except Exception as e:
log.debug(f"Failed to save escrow state: {e}")
@staticmethod
def _parse_funded_at(funded_at_str):
"""Parse funded_at string to naive local datetime (handles both UTC ISO and local)."""
dt = datetime.fromisoformat(funded_at_str)
if dt.tzinfo is not None:
# Convert UTC to local naive datetime
utc_ts = dt.timestamp()
dt = datetime.fromtimestamp(utc_ts)
return dt
def _check_stale_offers(self, name, platform):
"""Reduce the premium on offers that have gone unmatched.
Operates on the live offer list rather than local state, so manually created
offers are covered too.
Not included in the public repository — see the note at the top of this
module.
"""
raise NotImplementedError(
"_check_stale_offers is not part of the public repository")
def _record_trade(self, pname, contract):
currency = getattr(contract, 'currency', 'EUR')
try: spot_now = SpotPriceProvider.get_spot(currency)
except Exception as e: log.warning(f"Spot price for trade record: {e}"); spot_now = 0
btc = contract.amount_sats / 1e8
sell_price = contract.price_fiat
# Look up actual buy data from pending_escrows or cache
buy_data = {}
offer_id = getattr(contract, 'offer_id', '') or (contract.id.split('-')[0] if '-' in contract.id else '')
for oid in (offer_id, contract.id):
if oid:
# Check pending_escrows first, then cache
info = self.pending_escrows.get(oid, {})
buy_data = info.get("buy_data", {}) or self._buy_data_cache.pop(oid, {})
if buy_data:
break
if not buy_data and offer_id:
# Neither in memory nor cached: read the persisted record. If there is none, this may
# be a re-published offer whose record still sits under its old id — the contract's
# escrow address links the two even if the re-publish was never seen while live.
buy_data = self._escrow_state.get(str(offer_id), {}).get("buy_data") or {}
if not buy_data:
esc_addr = None
plat = self.platforms.get(pname)
if plat is not None and hasattr(plat, "get_contract_detail"):
try:
_esc = (plat.get_contract_detail(contract.id) or {}).get("escrow")
esc_addr = _esc if isinstance(_esc, str) else None
except Exception as e:
log.debug(f"{pname}: contract detail {contract.id}: {e}")
adopted = self._adopt_republished_state(offer_id, None, esc_addr)
buy_data = (adopted or {}).get("buy_data") or {}
# Kraken withdrawal fee: fixed 0.000015 BTC
KRAKEN_WITHDRAWAL_FEE_BTC = 0.000015
if buy_data.get("fiat_spent"):
spot_at_buy = buy_data.get("spot_at_buy", 0)
buy_currency = buy_data.get("buy_currency", currency)
funding_fee_sats = buy_data.get("funding_fee_sats", 0)
funding_fee = funding_fee_sats / 1e8 * spot_now if spot_now else 0
# Use proportional buy price: buyer may match fewer sats than full offer.
# Charge only the cost of the sats actually traded, not the full Kraken buy.
if spot_at_buy and spot_at_buy > 0:
# Effective buy cost = sats_traded * spot_at_buy (in buy_currency)
total_sats_bought = buy_data.get("btc_amount", 0) * 1e8 if buy_data.get("btc_amount") else 0
if total_sats_bought > 0 and contract.amount_sats < total_sats_bought * 0.99:
# Partial fill: prorate buy_price and fee
fill_ratio = contract.amount_sats / total_sats_bought
buy_price = buy_data["fiat_spent"] * fill_ratio
efee = buy_data.get("exchange_fee", 0) * fill_ratio
else:
buy_price = buy_data["fiat_spent"]
efee = buy_data.get("exchange_fee", 0)
else:
buy_price = buy_data["fiat_spent"]
efee = buy_data.get("exchange_fee", 0)
else:
# Fallback: estimate from current spot (old trades without buy_data)
# Determine actual exchange buy currency (may differ from sell currency)
ex = list(self.exchanges.values())[0] if self.exchanges else None
buy_currency = ex.get_fiat_currency() if ex and hasattr(ex, 'get_fiat_currency') else currency
try:
spot_buy = SpotPriceProvider.get_spot(buy_currency) if buy_currency != currency else spot_now
except Exception:
spot_buy = spot_now
buy_price = btc * spot_buy
efee = buy_price * 0.0026
spot_at_buy = spot_buy
# Estimate funding fee: ~150 vbytes × 5 sat/vB = 750 sats
funding_fee = 750 / 1e8 * spot_buy if spot_buy else 0
# Peach 2% fee is paid by the buyer, not the seller — excluded from our cost
# All fees in CHF (base accounting currency)
spot_chf = None
try: spot_chf = SpotPriceProvider.get_spot("CHF")
except Exception: pass
spot_for_fees = spot_at_buy if spot_at_buy else (spot_chf or spot_now)
withdrawal_fee = KRAKEN_WITHDRAWAL_FEE_BTC * spot_for_fees if spot_for_fees else 0
# Convert buy_price to CHF (base accounting currency).
# Profit model: buy and sell at same BTC spot; buyer pays buy_price × (1+premium%).
# For cross-currency trades, convert buy_price to CHF via current FX rate.
def _fx_to_chf(currency_code):
"""Get current CHF conversion rate for a fiat currency."""
if currency_code == "CHF": return 1.0
try:
pair = f"{currency_code}CHF" if currency_code != "USD" else "USDCHF"
r = _requests.get("https://api.kraken.com/0/public/Ticker",
params={"pair": pair}, timeout=5)
return float(list(r.json()["result"].values())[0]["c"][0])
except Exception:
return {"EUR": 0.94, "USD": 0.90}.get(currency_code, 1.0)
if buy_currency == "CHF":
buy_price_chf = buy_price
efee_chf = efee
else:
rate = _fx_to_chf(buy_currency)
buy_price_chf = buy_price * rate
efee_chf = efee * rate
# Get premium from pending_escrows (stored at offer creation)
oid = offer_id or (contract.id.split('-')[0] if '-' in contract.id else contract.id)
with self._escrow_lock:
pesc = self.pending_escrows.get(oid, {})
prem = pesc.get("premium") or self._escrow_state.get(str(oid), {}).get("premium") or 0
if not prem:
# Peach reports the premium on every contract — authoritative, nothing to derive.
prem = (getattr(contract, "raw_data", None) or {}).get("premium") or 0
if not prem:
# Last resort: derive it from the sale price. Both sides must be in the CONTRACT
# currency. spot_at_buy is in the BUY currency (EUR) while the sale price may be in
# CHF — mixing them turned a 6% trade into 0.13%.
prem = (sell_price / (btc * spot_now) - 1) * 100 if (btc and spot_now) else 0
# What the buyer ACTUALLY paid, converted to CHF.
#
# The old model assumed buy and sale happen at the same spot and booked
# buy_price × premium. But the marketplace prices a contract from its index AT THE
# MATCH, not when the offer was created — verified against its own contract data (a
# trade bought at 68,030 EUR/BTC sold against an index of 74,575 after waiting 74h,
# with the platform's premium matching ours exactly, so it was not a bookkeeping
# artefact). Support claims the price is locked from creation to sale; it is not. The
# move during the wait is real profit or loss: across 413 same-currency trades the
# model had left +615 EUR unbooked, with single trades between +84 and -26 EUR.
sell_currency = getattr(contract, "currency", currency)
fx_sell = 1.0 if sell_currency == "CHF" else _fx_to_chf(sell_currency)
sell_price_chf = sell_price * fx_sell
fees_chf = efee_chf + withdrawal_fee + funding_fee
net = sell_price_chf - buy_price_chf - fees_chf
# What the premium alone would have earned; the rest is the price move while waiting.
premium_only = buy_price_chf * (prem / 100) - fees_chf
self.trade_logger.log_trade(
platform=pname,
exchange=list(self.exchanges.keys())[0] if self.exchanges else "",
contract_id=contract.id, amount_sats=contract.amount_sats,
buy_price=buy_price, sell_price=sell_price,
currency=currency, buy_currency=buy_currency,
sell_price_chf=round(sell_price_chf, 2),
premium_pct=prem,
exchange_fee=efee, platform_fee=0,
network_fee=withdrawal_fee + funding_fee,
net_profit=round(net, 5), payment_method=contract.payment_method,
withdrawal_fee=withdrawal_fee, funding_fee=funding_fee,
spot_at_buy=spot_at_buy, spot_at_sell=spot_now)
self.daily_volume_sats += contract.amount_sats
log.info(f"{pname}: COMPLETE {contract.id[:12]} profit={net:.2f} CHF "
f"(buy={buy_price:.2f} {buy_currency} sell={sell_price:.2f} {sell_currency} "
f"prem={prem:.1f}% fees={fees_chf:.2f} price-move={net - premium_only:+.2f})")
def get_status(self):
uptime = datetime.now() - self._start_time
h, m = divmod(int(uptime.total_seconds()) // 60, 60)
d, h = divmod(h, 24)
uptime_str = f"{d}d {h}h {m}m" if d else f"{h}h {m}m"
return {"paused":self.paused,"uptime":uptime_str,"daily_volume_sats":self.daily_volume_sats,"pending_escrows":len(self.pending_escrows),"funded_escrows":sum(1 for e in self.pending_escrows.values() if e.get("funded")),"platforms":{n:p.get_status() for n,p in self.platforms.items()},"exchanges":{n:e.get_status() for n,e in self.exchanges.items()}}