lupier/scalable-wealth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update scalable_sync.py to add SC_PORTFOLIO_ID environment variable handling and ensure broker context is set before syncing transactions

Browse Source
This commit is contained in:
Lukas Pierchalla
2026-09-20 13:02:29 +02:00
parent 3004c69234
commit 4f38be33d7
1 changed files with 71 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scalable_sync.py
+71 -39
View File
@@ -34,6 +34,7 @@ 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", "{}"))
SC_PORTFOLIO_ID = os.getenv("SC_PORTFOLIO_ID")
DATA_DIR = Path("/app/data")
@@ -63,27 +64,65 @@ def sc(*args: str, timeout: int = 60) -> dict:
return json.loads(result.stdout)
def sc_ensure_context():
"""
Check if a broker context is set. If not (or if SC_PORTFOLIO_ID is configured),
call 'sc broker context select' to set it. Falls back to 'sc login' if needed.
"""
# If a portfolio ID is explicitly configured, always select it
if SC_PORTFOLIO_ID:
log.info("Setting broker context to portfolio %s", SC_PORTFOLIO_ID)
try:
sc("broker", "context", "select", "--portfolio-id", SC_PORTFOLIO_ID)
log.info("Broker context set via SC_PORTFOLIO_ID")
return
except Exception as e:
log.warning("Failed to set context via SC_PORTFOLIO_ID: %s", e)
# Check if context already exists
try:
ctx_raw = sc("broker", "context", "show")
ctx = ctx_raw.get("result", ctx_raw) if isinstance(ctx_raw, dict) else {}
portfolio_id = ctx.get("portfolio_id") or ctx_raw.get("portfolio_id")
if portfolio_id:
log.info("Broker context already set: portfolio=%s", portfolio_id)
return
except Exception:
pass
# No context — need to login
log.info("No broker context found, attempting sc login...")
env = {**os.environ, "HOME": "/home/appuser", "XDG_CONFIG_HOME": "/home/appuser/.config"}
result = subprocess.run(["sc", "login"], env=env, timeout=120)
if result.returncode != 0:
raise RuntimeError("sc login failed — cannot proceed without broker context")
log.info("sc login completed")
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 from_time:
args += ["--from-time", from_time.strftime("%Y-%m-%dT%H:%M:%SZ")]
if to_time:
args += ["--to-time", to_time.strftime("%Y-%m-%dT%H:%M:%SZ")]
if type_filter:
for t in type_filter: args += ["--type-filter", t]
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]
for s in status:
args += ["--status", s]
if isin:
args += ["--isin", isin]
if cursor:
args += ["--cursor", cursor]
raw = sc(*args)
result = raw.get("result", raw)
return {"items": result.get("items", []), "cursor": result.get("cursor")}
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")
raw = sc("broker", "holdings")
return raw.get("result", raw)
# ---------------------------------------------------------------------------
@@ -173,18 +212,17 @@ def audit(action, payload):
def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]:
tx_type = tx.get("type") # "SECURITY_TRANSACTION" | "CASH_TRANSACTION"
tx_type = tx.get("type")
status = (tx.get("status") or "").upper()
if tx.get("is_cancellation"):
return []
if SC_SETTLED_ONLY and status != "SETTLED":
return []
last_event = (tx.get("last_event_datetime") or "")[:10]
tx_id = tx.get("id", "")
amount = abs(tx.get("amount") or 0)
if tx.get("is_cancellation"):
return []
if SC_SETTLED_ONLY and status not in {"SETTLED", "CONFIRMED"}:
return []
activities = []
if tx_type == "SECURITY_TRANSACTION":
@@ -194,10 +232,8 @@ def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]:
dead_letter(tx, f"unmapped ISIN: {isin}")
return []
side = (tx.get("side") or "").upper() # "BUY" | "SELL"
side = (tx.get("side") or "").upper()
qty = tx.get("quantity")
# unit_price aus amount/quantity ableiten
unit_price = round(amount / qty, 6) if qty else None
activities.append(WealthfolioActivity(
@@ -207,8 +243,6 @@ def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]:
quantity=qty,
unit_price=unit_price,
amount=amount,
fee=None,
tax=None,
external_id=tx_id,
account_id=WF_ACCOUNT_ID,
))
@@ -222,8 +256,8 @@ def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]:
"FEE": "FEE",
"TAX": "TAX",
"TAX_RETURN": "CREDIT",
"CASH_TRANSFER_IN": "DEPOSIT", # neu
"CASH_TRANSFER_OUT": "WITHDRAWAL", # neu
"CASH_TRANSFER_IN": "DEPOSIT",
"CASH_TRANSFER_OUT": "WITHDRAWAL",
}.get(cash_type)
if not wf_type:
@@ -234,7 +268,7 @@ def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]:
date=last_event,
symbol="$CASH",
activity_type=wf_type,
amount=abs(tx.get("amount") or 0),
amount=amount,
external_id=tx_id,
account_id=WF_ACCOUNT_ID,
))
@@ -312,13 +346,15 @@ class WealthfolioMCP:
}
for activity in activities
]
result = asyncio.run(
self._connect_and_call("record_activities", {"activities": payload})
)
log.info("MCP record_activities result: %s", result) # ← neu
if getattr(result, "is_error", False):
raise RuntimeError(f"record_activities error: {result.content}")
log.info("record_activities OK: %s activities", len(activities))
return result
# ---------------------------------------------------------------------------
# Sync
# ---------------------------------------------------------------------------
@@ -336,11 +372,11 @@ def reconcile(mcp):
def sync_once(dry_run=False):
init_db()
mcp = None
sc_ensure_context()
mcp = None
if not dry_run:
mcp = WealthfolioMCP(WF_MCP_URL, WF_MCP_TOKEN)
if not mcp.ping():
raise RuntimeError("Wealthfolio MCP unreachable")
@@ -351,8 +387,8 @@ def sync_once(dry_run=False):
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")
items = page["items"]
cursor = page["cursor"]
for summary in items:
tx_id = summary.get("id")
@@ -365,9 +401,6 @@ def sync_once(dry_run=False):
continue
activities = map_to_wealthfolio(summary)
if activities:
all_activities.extend(activities)
mark_imported(...)
if not activities:
stats["dead_letter"] += 1
continue
@@ -377,18 +410,17 @@ def sync_once(dry_run=False):
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]})
stats["new"] += len(activities) # ← diese Zeile fehlt noch
stats["new"] += len(activities)
else:
try:
result = mcp.record_activities(activities)
mcp.record_activities(activities)
mark_imported(tx_id, tx_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 all_activities and not dry_run:
mcp.record_activities(all_activities)
if not cursor or not items:
break