initial - ai

Dockerfile

@@ -0,0 +1,41 @@
+# syntax=docker/dockerfile:1
+FROM python:3.13-slim-bookworm AS builder
+
+ARG SCALABLE_CLI_VERSION=0.1.0
+ARG TARGETARCH
+
+RUN apt-get update && apt-get install -y --no-install-recommends curl ca-certificates \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN ARCH=$([ "$TARGETARCH" = "arm64" ] && echo "aarch64" || echo "x86_64") \
+    && curl -fsSL -o /tmp/sc.tar.gz \
+       "https://github.com/ScalableCapital/scalable-cli/releases/download/v${SCALABLE_CLI_VERSION}/scalable-cli-${ARCH}-unknown-linux-gnu.tar.gz" \
+    && curl -fsSL -o /tmp/sc.tar.gz.sha256 \
+       "https://github.com/ScalableCapital/scalable-cli/releases/download/v${SCALABLE_CLI_VERSION}/scalable-cli-${ARCH}-unknown-linux-gnu.tar.gz.sha256" \
+    && cd /tmp && sha256sum -c sc.tar.gz.sha256 \
+    && tar -xzf sc.tar.gz \
+    && mv scalable-cli-${ARCH}-unknown-linux-gnu/sc /usr/local/bin/sc \
+    && chmod +x /usr/local/bin/sc
+
+FROM python:3.13-slim-bookworm
+
+ARG PUID=1000
+ARG PGID=1000
+
+# User mit konfigurierbarer UID/GID anlegen
+RUN groupadd -g ${PGID} appgroup \
+    && useradd -u ${PUID} -g appgroup -m -d /home/appuser appuser
+
+COPY --from=builder /usr/local/bin/sc /usr/local/bin/sc
+
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY scalable_sync.py .
+
+ENV SC_CONFIG_DIR=/tmp/sc-config
+
+USER ${PUID}:${PGID}
+
+ENTRYPOINT ["python", "/app/scalable_sync.py"]

README.md

@@ -1,2 +1,111 @@
-# scalable-wealth
+# scalable-sync
 
+Synchronisiert Scalable Capital Broker-Transaktionen nach Wealthfolio.
+
+## Voraussetzungen
+
+- Docker + Docker Compose
+- Wealthfolio mit MCP aktiviert (`WF_MCP_ENABLED=true`)
+- Scalable Capital: Agentic Investing aktiviert (Profil → Sicherheit)
+
+## Dateien
+
+| Datei | Zweck |
+|---|---|
+| `Dockerfile` | Container-Image |
+| `compose.yaml` | Docker Compose Config |
+| `.env` | Alle Einstellungen |
+| `scalable_sync.py` | Sync-Adapter |
+| `requirements.txt` | Python-Abhängigkeiten |
+
+## Setup
+
+```bash
+# Dateien in ein Verzeichnis deiner Wahl kopieren, z. B.:
+# /opt/scalable-sync/ oder ~/docker/scalable-sync/
+
+docker compose build
+
+# 1. Einmalig einloggen (Session wird im Volume gespeichert)
+docker compose run --rm -it scalable-sync sc login --local-read-only
+
+# 2. Test
+docker compose run --rm scalable-sync sc whoami
+docker compose run --rm scalable-sync sc broker holdings --json | jq .
+
+# 3. Dry-Run
+docker compose run --rm scalable-sync python /app/scalable_sync.py --once --dry-run
+
+# 4. Start
+docker compose up -d
+```
+
+## .env
+
+```bash
+# Wealthfolio
+WF_MCP_URL=http://wealthfolio:8088/mcp
+WF_MCP_TOKEN=wf_mcp_dein_token_hier
+WF_ACCOUNT_ID=dein-konto-uuid
+
+# Optional: Scalable Portfolio-ID (bei mehreren Portfolios)
+SC_PORTFOLIO_ID=
+
+# Sync
+SYNC_INTERVAL_HOURS=6
+SC_OVERLAP_DAYS=14
+SC_SETTLED_ONLY=true
+
+# ISIN → Wealthfolio-Symbol
+ASSET_MAP={"IE00B4L5Y983":"EUNL.DE","IE00B3RBWM25":"VWCE.DE"}
+
+# Container-User
+PUID=1000
+PGID=1000
+```
+
+### WF_ACCOUNT_ID finden
+
+Wealthfolio → Accounts → Konto anklicken → ID aus der URL kopieren (`/accounts/abc123-def456/...`).
+
+### ASSET_MAP
+
+JSON-Objekt: `"ISIN": "SYMBOL"`. Wealthfolio-Symbol unter Assets suchen.
+
+### SC_PORTFOLIO_ID
+
+Nur bei mehreren Scalable-Portfolios nötig. Ermitteln mit:
+```bash
+docker compose run --rm scalable-sync sc broker context show --json
+```
+
+## Volumes
+
+Docker legt diese Volumes automatisch an — kein manuelles `mkdir` nötig.
+
+| Volume | Inhalt | Zweck |
+|---|---|---|
+| `scalable-sync-data` | state.db, audit.jsonl, dead-letter.jsonl | Sync-Status und Logs |
+| `scalable-sync-config` | config.toml, session.json | Scalable CLI Session |
+
+**Wichtig:** Ohne `scalable-sync-config` geht die Session bei `docker compose down` verloren.
+
+## Befehle
+
+| Befehl | Beschreibung |
+|---|---|
+| `docker compose run --rm scalable-sync python /app/scalable_sync.py --once --dry-run` | Testlauf ohne Schreiben |
+| `docker compose run --rm scalable-sync python /app/scalable_sync.py --once` | Einmaliger Sync |
+| `docker compose logs -f` | Logs anzeigen |
+| `docker compose down && docker compose up -d` | Neustart (Session bleibt erhalten) |
+
+## Troubleshooting
+
+| Problem | Lösung |
+|---|---|
+| "Not logged in" | `docker compose run --rm -it scalable-sync sc login --local-read-only` |
+| Keine Transaktionen | `SC_SETTLED_ONLY` und `SC_OVERLAP_DAYS` prüfen |
+| Unmapped ISIN | ISIN in `ASSET_MAP` ergänzen |
+| Mehrere Portfolios | `SC_PORTFOLIO_ID` setzen |
+| Doppelte Activities | Prüfen ob `WF_ACCOUNT_ID` korrekt ist |
+| Session nach Neustart weg | `scalable-sync-config` Volume in compose.yaml prüfen |

compose.yaml

@@ -0,0 +1,34 @@
+services:
+  scalable-sync:
+    build:
+      context: .
+      args:
+        SCALABLE_CLI_VERSION: "0.1.0"
+        PUID: ${PUID}
+        PGID: ${PGID}
+    image: scalable-sync:latest
+    hostname: scalable-sync
+    container_name: scalable-sync
+    restart: unless-stopped
+    user: "${PUID}:${PGID}"
+    networks:
+      - ext_net
+    volumes:
+      - scalable-sync-data:/app/data
+      - scalable-sync-config:/tmp/sc-config
+    env_file:
+      - .env
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request,json,os; req=urllib.request.Request(os.environ['WF_MCP_URL'],method='POST',data=json.dumps({'jsonrpc':'2.0','id':1,'method':'ping'}).encode(),headers={'Authorization':'Bearer '+os.environ['WF_MCP_TOKEN'],'Content-Type':'application/json'}); urllib.request.urlopen(req,timeout=5)"]
+      interval: 5m
+      timeout: 10s
+      retries: 3
+      start_period: 30s
+
+networks:
+  ext_net:
+    external: true
+
+volumes:
+  scalable-sync-data:
+  scalable-sync-config:

example.env

@@ -0,0 +1,16 @@
+# ── Wealthfolio ──────────────────────────────────────────────
+WF_MCP_URL=http://wealthfolio:8088/mcp
+WF_MCP_TOKEN=wf_mcp_dein_token_hier
+
+# ── Sync-Intervalle ──────────────────────────────────────────
+SYNC_INTERVAL_HOURS=6
+SC_OVERLAP_DAYS=14
+SC_SETTLED_ONLY=true
+SC_MAPPING_VERSION=1
+
+# ── Asset-Mapping ──────────
+ASSET_MAP={"IE00B4L5Y983":"EUNL.DE","IE00B3RBWM25":"VWCE.DE","DE0002635307":"EXSA.DE"}
+
+# ── User im Container ─────────
+PUID=1000
+PGID=1000

requirements.txt

@@ -0,0 +1 @@
+requests>=2.32.0

scalable_sync.py

@@ -0,0 +1,398 @@
+#!/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 tempfile
+import time
+from dataclasses import dataclass, asdict
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Optional
+
+import requests
+
+# ---------------------------------------------------------------------------
+# 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", "{}"))
+
+# Scalable CLI Config: tmpfs-basiert, automatisch generiert
+SC_CONFIG_DIR = Path(os.getenv("SC_CONFIG_DIR", tempfile.mkdtemp(prefix="sc-config-")))
+SC_CONFIG_DIR.mkdir(parents=True, exist_ok=True)
+(SC_CONFIG_DIR / "config.toml").write_text(
+    '[auth]\nsession_backend = "file"\nsigning_key_backend = "file"\n'
+)
+os.environ["SC_CONFIG_DIR"] = str(SC_CONFIG_DIR)
+
+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/sync", "SC_CONFIG_DIR": str(SC_CONFIG_DIR)}
+    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, token):
+        self.url = url
+        self.session = requests.Session()
+        self.session.headers.update({
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        })
+
+    def _rpc(self, method, params=None):
+        payload = {"jsonrpc": "2.0", "id": int(time.time() * 1000), "method": method}
+        if params: payload["params"] = params
+        resp = self.session.post(self.url, json=payload, timeout=30)
+        resp.raise_for_status()
+        return resp.json()
+
+    def ping(self):
+        try:
+            return self._rpc("ping").get("result") == "pong"
+        except Exception as e:
+            log.error("MCP ping failed: %s", e)
+            return False
+
+    def record_activities(self, activities):
+        payload = [{
+            "accountId": a.account_id, "date": a.date, "symbol": a.symbol, "activityType": a.activity_type,
+            "quantity": a.quantity, "unitPrice": a.unit_price, "amount": a.amount,
+            "fee": a.fee, "tax": a.tax, "currency": a.currency,
+            "source": a.source, "externalId": a.external_id,
+        } for a in activities]
+        return self._rpc("tools/call", {"name": "record_activities", "arguments": {"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 = 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
+
+    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()