#!/usr/bin/env python3 """ scalable-sync: Scalable Capital → Wealthfolio MCP Sync Adapter Alle Config über Environment-Variablen. """ import json import logging import os import sqlite3 import subprocess import sys import time from dataclasses import dataclass, asdict from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Optional import asyncio import httpx from mcp import ClientSession from mcp.client.streamable_http import streamable_http_client # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- WF_MCP_URL = os.environ["WF_MCP_URL"] WF_MCP_TOKEN = os.environ["WF_MCP_TOKEN"] WF_ACCOUNT_ID = os.environ.get("WF_ACCOUNT_ID") SYNC_INTERVAL_HOURS = int(os.getenv("SYNC_INTERVAL_HOURS", "6")) SC_OVERLAP_DAYS = int(os.getenv("SC_OVERLAP_DAYS", "14")) SC_SETTLED_ONLY = os.getenv("SC_SETTLED_ONLY", "true").lower() == "true" SC_MAPPING_VERSION = int(os.getenv("SC_MAPPING_VERSION", "1")) ASSET_MAP = json.loads(os.getenv("ASSET_MAP", "{}")) DATA_DIR = Path("/app/data") DATA_DIR.mkdir(exist_ok=True) DB_PATH = DATA_DIR / "state.db" AUDIT_PATH = DATA_DIR / "audit.jsonl" DEAD_LETTER_PATH = DATA_DIR / "dead-letter.jsonl" logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", handlers=[logging.StreamHandler(sys.stdout)] ) log = logging.getLogger("scalable-sync") # --------------------------------------------------------------------------- # Scalable CLI Wrapper # --------------------------------------------------------------------------- def sc(*args: str, timeout: int = 60) -> dict: cmd = ["sc"] + list(args) env = {**os.environ, "HOME": "/home/appuser", "XDG_CONFIG_HOME": "/home/appuser/.config"} result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, env=env) if result.returncode != 0: log.error("sc failed: %s", result.stderr) raise RuntimeError(f"sc exit {result.returncode}: {result.stderr}") return json.loads(result.stdout) def sc_transactions(from_time=None, to_time=None, type_filter=None, status=None, isin=None, cursor=None, page_size=100): args = ["broker", "transactions", "--page-size", str(page_size)] if from_time: args += ["--from-time", from_time.isoformat()] if to_time: args += ["--to-time", to_time.isoformat()] if type_filter: for t in type_filter: args += ["--type-filter", t] if status: for s in status: args += ["--status", s] if isin: args += ["--isin", isin] if cursor: args += ["--cursor", cursor] return sc(*args) def sc_transaction_details(tx_id: str) -> dict: return sc("broker", "transaction", "details", "--transaction-id", tx_id) def sc_holdings() -> dict: return sc("broker", "holdings") # --------------------------------------------------------------------------- # State # --------------------------------------------------------------------------- def init_db(): with sqlite3.connect(DB_PATH) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS sync_state ( scalable_transaction_id TEXT PRIMARY KEY, source_hash TEXT NOT NULL, source_status TEXT NOT NULL, wealthfolio_activity_ids TEXT, imported_at TEXT, last_seen_at TEXT NOT NULL, mapping_version INTEGER NOT NULL ) """) conn.execute("CREATE INDEX IF NOT EXISTS idx_last_seen ON sync_state(last_seen_at)") def hash_transaction(tx: dict) -> str: relevant = {k: tx.get(k) for k in [ "id", "type", "status", "last_event_datetime", "security", "security_trade", "cash", "non_trade_security" ]} return json.dumps(relevant, sort_keys=True, default=str) def is_imported(tx_id: str, current_hash: str) -> bool: with sqlite3.connect(DB_PATH) as conn: row = conn.execute("SELECT source_hash FROM sync_state WHERE scalable_transaction_id = ?", (tx_id,)).fetchone() return row is not None and row[0] == current_hash def mark_imported(tx_id, tx_hash, status, wf_ids): now = datetime.now(timezone.utc).isoformat() with sqlite3.connect(DB_PATH) as conn: conn.execute(""" INSERT OR REPLACE INTO sync_state VALUES (?, ?, ?, ?, ?, ?, ?) """, (tx_id, tx_hash, status, json.dumps(wf_ids), now, now, SC_MAPPING_VERSION)) def mark_seen(tx_id, tx_hash, status): now = datetime.now(timezone.utc).isoformat() with sqlite3.connect(DB_PATH) as conn: conn.execute(""" INSERT OR IGNORE INTO sync_state (scalable_transaction_id, source_hash, source_status, last_seen_at, mapping_version) VALUES (?, ?, ?, ?, ?) """, (tx_id, tx_hash, status, now, SC_MAPPING_VERSION)) conn.execute("UPDATE sync_state SET last_seen_at = ? WHERE scalable_transaction_id = ?", (now, tx_id)) # --------------------------------------------------------------------------- # Mapping # --------------------------------------------------------------------------- @dataclass class WealthfolioActivity: date: str symbol: str activity_type: str quantity: Optional[float] = None unit_price: Optional[float] = None amount: Optional[float] = None fee: Optional[float] = None tax: Optional[float] = None currency: str = "EUR" source: str = "scalable-cli" external_id: Optional[str] = None account_id: Optional[str] = None def dead_letter(tx, reason): entry = {"timestamp": datetime.now(timezone.utc).isoformat(), "reason": reason, "transaction": tx} with open(DEAD_LETTER_PATH, "a") as f: f.write(json.dumps(entry, default=str) + "\n") log.warning("Dead letter: %s (%s)", tx.get("id"), reason) def audit(action, payload): entry = {"timestamp": datetime.now(timezone.utc).isoformat(), "action": action, "payload": payload} with open(AUDIT_PATH, "a") as f: f.write(json.dumps(entry, default=str) + "\n") def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]: tx_type = tx.get("type") status = (tx.get("status") or "").upper() if tx.get("is_cancellation") or status in {"CANCELLED", "EXPIRED", "REJECTED", "CANCEL_REQUESTED"}: return [] if SC_SETTLED_ONLY and status not in {"SETTLED", "CONFIRMED"}: return [] if tx.get("is_pending"): return [] sec = tx.get("security") or {} isin = sec.get("isin") symbol = ASSET_MAP.get(isin) if isin else None st = tx.get("security_trade") or {} cash = tx.get("cash") or {} nts = tx.get("non_trade_security") or {} last_event = (tx.get("last_event_datetime") or "")[:10] tx_id = tx.get("id", "") activities = [] if tx_type in {"BUY", "SELL", "SAVINGS_PLAN", "REINVESTMENT"}: if not symbol: dead_letter(tx, f"unmapped ISIN: {isin}") return [] qty = (st.get("number_of_shares") or {}).get("filled") or nts.get("quantity") price = st.get("average_price") or nts.get("average_price") amount = st.get("total_amount") or nts.get("total_amount") fee = st.get("fee") or st.get("transactional_fee") tax = (st.get("aggregated_transaction_taxes") or {}).get("total_tax") activities.append(WealthfolioActivity( date=last_event, symbol=symbol, activity_type="BUY" if tx_type in {"BUY", "SAVINGS_PLAN", "REINVESTMENT"} else "SELL", quantity=qty, unit_price=price, amount=amount, fee=fee, tax=tax, external_id=tx_id, account_id=WF_ACCOUNT_ID )) elif tx_type == "DEPOSIT": activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="DEPOSIT", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) elif tx_type == "WITHDRAWAL": activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="WITHDRAWAL", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) elif tx_type == "DISTRIBUTION": if not symbol: dead_letter(tx, "distribution without ISIN") return [] activities.append(WealthfolioActivity(date=last_event, symbol=symbol, activity_type="DIVIDEND", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) elif tx_type == "INTEREST": activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="INTEREST", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) elif tx_type == "FEE": activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="FEE", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) elif tx_type == "TAX": activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="TAX", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) elif tx_type == "TAX_RETURN": activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="CREDIT", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) elif tx_type in {"TRANSFER_IN", "CASH_TRANSFER_IN"}: target = "TRANSFER_IN" if symbol else "DEPOSIT" activities.append(WealthfolioActivity( date=last_event, symbol=symbol or "$CASH", activity_type=target, quantity=nts.get("quantity"), amount=cash.get("amount") or nts.get("total_amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID )) elif tx_type in {"TRANSFER_OUT", "CASH_TRANSFER_OUT"}: target = "TRANSFER_OUT" if symbol else "WITHDRAWAL" activities.append(WealthfolioActivity( date=last_event, symbol=symbol or "$CASH", activity_type=target, quantity=nts.get("quantity"), amount=cash.get("amount") or nts.get("total_amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID )) elif tx_type in { "SWAP_IN", "SWAP_OUT", "CURRENCY_SWITCH_BUY", "CURRENCY_SWITCH_SELL", "POCKET_MONEY", "REINVESTMENT_DISTRIBUTION", "REINVESTMENT_POCKET_MONEY" }: dead_letter(tx, f"unsupported type: {tx_type}") return [] else: dead_letter(tx, f"unknown type: {tx_type}") return [] return activities # --------------------------------------------------------------------------- # Wealthfolio MCP Client # --------------------------------------------------------------------------- class WealthfolioMCP: def __init__(self, url: str, token: str): self.url = url self.token = token def _http_client(self) -> httpx.AsyncClient: return httpx.AsyncClient( headers={ "Authorization": f"Bearer {self.token}", "Accept": "application/json, text/event-stream", }, timeout=30.0, ) async def _connect_and_call(self, tool_name: str, arguments: dict): async with self._http_client() as http_client: async with streamable_http_client( self.url, http_client=http_client, ) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: await session.initialize() return await session.call_tool(tool_name, arguments) async def _list_tools(self) -> list[str]: async with self._http_client() as http_client: async with streamable_http_client( self.url, http_client=http_client, ) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: await session.initialize() result = await session.list_tools() return [tool.name for tool in result.tools] def ping(self) -> bool: try: tools = asyncio.run(self._list_tools()) log.info("Available Wealthfolio MCP tools: %s", tools) return True except Exception as exc: log.exception("MCP initialization failed: %s", exc) return False def record_activities(self, activities): payload = [ { "accountId": activity.account_id, "date": activity.date, "symbol": activity.symbol, "activityType": activity.activity_type, "quantity": activity.quantity, "unitPrice": activity.unit_price, "amount": activity.amount, "fee": activity.fee, "tax": activity.tax, "currency": activity.currency, "source": activity.source, "externalId": activity.external_id, } for activity in activities ] return asyncio.run( self._connect_and_call( "record_activities", {"activities": payload}, ) ) # --------------------------------------------------------------------------- # Sync # --------------------------------------------------------------------------- def reconcile(mcp): try: sc_h = sc_holdings() log.info("Scalable holdings count: %s", len(sc_h.get("items", []))) audit("reconcile", {"scalable_holdings": sc_h}) return True except Exception as e: log.error("Reconcile failed: %s", e) return False def sync_once(dry_run=False): init_db() mcp = None if not dry_run: mcp = WealthfolioMCP(WF_MCP_URL, WF_MCP_TOKEN) if not mcp.ping(): raise RuntimeError("Wealthfolio MCP unreachable") from_time = datetime.now(timezone.utc) - timedelta(days=SC_OVERLAP_DAYS) to_time = datetime.now(timezone.utc) stats = {"new": 0, "skipped": 0, "dead_letter": 0, "error": 0} cursor = None while True: page = sc_transactions(from_time=from_time, to_time=to_time, cursor=cursor, page_size=100) items = page.get("items", []) cursor = page.get("cursor") for summary in items: tx_id = summary.get("id") tx_hash = hash_transaction(summary) status = (summary.get("status") or "").upper() if is_imported(tx_id, tx_hash): mark_seen(tx_id, tx_hash, status) stats["skipped"] += 1 continue try: detail = sc_transaction_details(tx_id) except Exception as e: log.error("Failed details %s: %s", tx_id, e) stats["error"] += 1 continue detail_hash = hash_transaction(detail) activities = map_to_wealthfolio(detail) if not activities: stats["dead_letter"] += 1 continue if dry_run: log.info("[DRY RUN] Would import %s: %s", tx_id, [ f"{a.activity_type} {a.symbol} qty={a.quantity} amt={a.amount}" for a in activities ]) audit("dry_run", {"tx_id": tx_id, "activities": [asdict(a) for a in activities]}) else: try: result = mcp.record_activities(activities) mark_imported(tx_id, detail_hash, status, [str(i) for i in range(len(activities))]) audit("imported", {"tx_id": tx_id, "activities": len(activities)}) stats["new"] += len(activities) except Exception as e: log.error("Import failed %s: %s", tx_id, e) stats["error"] += 1 if not cursor or not items: break if not dry_run: reconcile(mcp) log.info("Sync complete: %s", stats) return stats def main(): import argparse p = argparse.ArgumentParser() p.add_argument("--once", action="store_true") p.add_argument("--dry-run", action="store_true") p.add_argument("--daemon", action="store_true") args = p.parse_args() if args.once or args.dry_run: stats = sync_once(dry_run=args.dry_run) sys.exit(0 if stats["error"] == 0 else 1) if args.daemon: log.info("Starting daemon, interval=%dh", SYNC_INTERVAL_HOURS) while True: try: sync_once() except Exception as e: log.exception("Sync failed: %s", e) time.sleep(SYNC_INTERVAL_HOURS * 3600) else: p.print_help() if __name__ == "__main__": main()