update scalable_sync.py to enhance transaction mapping logic and improve handling of cash transactions
This commit is contained in:
+54
-69
@@ -73,7 +73,9 @@ def sc_transactions(from_time=None, to_time=None, type_filter=None, status=None,
|
|||||||
for s in status: args += ["--status", s]
|
for s in status: args += ["--status", s]
|
||||||
if isin: args += ["--isin", isin]
|
if isin: args += ["--isin", isin]
|
||||||
if cursor: args += ["--cursor", cursor]
|
if cursor: args += ["--cursor", cursor]
|
||||||
return sc(*args)
|
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:
|
def sc_transaction_details(tx_id: str) -> dict:
|
||||||
@@ -106,8 +108,9 @@ def init_db():
|
|||||||
|
|
||||||
def hash_transaction(tx: dict) -> str:
|
def hash_transaction(tx: dict) -> str:
|
||||||
relevant = {k: tx.get(k) for k in [
|
relevant = {k: tx.get(k) for k in [
|
||||||
"id", "type", "status", "last_event_datetime", "security",
|
"id", "type", "status", "last_event_datetime",
|
||||||
"security_trade", "cash", "non_trade_security"
|
"side", "security_transaction_type", "cash_transaction_type",
|
||||||
|
"amount", "quantity", "isin",
|
||||||
]}
|
]}
|
||||||
return json.dumps(relevant, sort_keys=True, default=str)
|
return json.dumps(relevant, sort_keys=True, default=str)
|
||||||
|
|
||||||
@@ -170,80 +173,70 @@ def audit(action, payload):
|
|||||||
|
|
||||||
|
|
||||||
def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]:
|
def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]:
|
||||||
tx_type = tx.get("type")
|
tx_type = tx.get("type") # "SECURITY_TRANSACTION" | "CASH_TRANSACTION"
|
||||||
status = (tx.get("status") or "").upper()
|
status = (tx.get("status") or "").upper()
|
||||||
if tx.get("is_cancellation") or status in {"CANCELLED", "EXPIRED", "REJECTED", "CANCEL_REQUESTED"}:
|
|
||||||
|
if tx.get("is_cancellation"):
|
||||||
return []
|
return []
|
||||||
if SC_SETTLED_ONLY and status not in {"SETTLED", "CONFIRMED"}:
|
if SC_SETTLED_ONLY and status != "SETTLED":
|
||||||
return []
|
|
||||||
if tx.get("is_pending"):
|
|
||||||
return []
|
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]
|
last_event = (tx.get("last_event_datetime") or "")[:10]
|
||||||
tx_id = tx.get("id", "")
|
tx_id = tx.get("id", "")
|
||||||
|
amount = abs(tx.get("amount") or 0)
|
||||||
|
|
||||||
activities = []
|
activities = []
|
||||||
|
|
||||||
if tx_type in {"BUY", "SELL", "SAVINGS_PLAN", "REINVESTMENT"}:
|
if tx_type == "SECURITY_TRANSACTION":
|
||||||
|
isin = tx.get("isin")
|
||||||
|
symbol = ASSET_MAP.get(isin) if isin else None
|
||||||
if not symbol:
|
if not symbol:
|
||||||
dead_letter(tx, f"unmapped ISIN: {isin}")
|
dead_letter(tx, f"unmapped ISIN: {isin}")
|
||||||
return []
|
return []
|
||||||
qty = (st.get("number_of_shares") or {}).get("filled") or nts.get("quantity")
|
|
||||||
price = st.get("average_price") or nts.get("average_price")
|
side = (tx.get("side") or "").upper() # "BUY" | "SELL"
|
||||||
amount = st.get("total_amount") or nts.get("total_amount")
|
qty = tx.get("quantity")
|
||||||
fee = st.get("fee") or st.get("transactional_fee")
|
|
||||||
tax = (st.get("aggregated_transaction_taxes") or {}).get("total_tax")
|
# unit_price aus amount/quantity ableiten
|
||||||
|
unit_price = round(amount / qty, 6) if qty else None
|
||||||
|
|
||||||
activities.append(WealthfolioActivity(
|
activities.append(WealthfolioActivity(
|
||||||
date=last_event, symbol=symbol,
|
date=last_event,
|
||||||
activity_type="BUY" if tx_type in {"BUY", "SAVINGS_PLAN", "REINVESTMENT"} else "SELL",
|
symbol=symbol,
|
||||||
quantity=qty, unit_price=price, amount=amount, fee=fee, tax=tax,
|
activity_type="BUY" if side == "BUY" else "SELL",
|
||||||
external_id=tx_id, account_id=WF_ACCOUNT_ID
|
quantity=qty,
|
||||||
|
unit_price=unit_price,
|
||||||
|
amount=amount,
|
||||||
|
fee=None,
|
||||||
|
tax=None,
|
||||||
|
external_id=tx_id,
|
||||||
|
account_id=WF_ACCOUNT_ID,
|
||||||
))
|
))
|
||||||
|
|
||||||
elif tx_type == "DEPOSIT":
|
elif tx_type == "CASH_TRANSACTION":
|
||||||
activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="DEPOSIT", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID))
|
cash_type = (tx.get("cash_transaction_type") or "").upper()
|
||||||
elif tx_type == "WITHDRAWAL":
|
wf_type = {
|
||||||
activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="WITHDRAWAL", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID))
|
"DEPOSIT": "DEPOSIT",
|
||||||
elif tx_type == "DISTRIBUTION":
|
"WITHDRAWAL": "WITHDRAWAL",
|
||||||
if not symbol:
|
"INTEREST": "INTEREST",
|
||||||
dead_letter(tx, "distribution without ISIN")
|
"FEE": "FEE",
|
||||||
|
"TAX": "TAX",
|
||||||
|
"TAX_RETURN": "CREDIT",
|
||||||
|
}.get(cash_type)
|
||||||
|
|
||||||
|
if not wf_type:
|
||||||
|
dead_letter(tx, f"unknown cash_transaction_type: {cash_type}")
|
||||||
return []
|
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(
|
activities.append(WealthfolioActivity(
|
||||||
date=last_event, symbol=symbol or "$CASH", activity_type=target,
|
date=last_event,
|
||||||
quantity=nts.get("quantity"), amount=cash.get("amount") or nts.get("total_amount"),
|
symbol="$CASH",
|
||||||
external_id=tx_id, account_id=WF_ACCOUNT_ID
|
activity_type=wf_type,
|
||||||
|
amount=abs(tx.get("amount") or 0),
|
||||||
|
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:
|
else:
|
||||||
dead_letter(tx, f"unknown type: {tx_type}")
|
dead_letter(tx, f"unknown type: {tx_type}")
|
||||||
return []
|
return []
|
||||||
@@ -370,15 +363,7 @@ def sync_once(dry_run=False):
|
|||||||
stats["skipped"] += 1
|
stats["skipped"] += 1
|
||||||
continue
|
continue
|
||||||
|
|
||||||
try:
|
activities = map_to_wealthfolio(summary)
|
||||||
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:
|
if not activities:
|
||||||
stats["dead_letter"] += 1
|
stats["dead_letter"] += 1
|
||||||
continue
|
continue
|
||||||
@@ -391,7 +376,7 @@ def sync_once(dry_run=False):
|
|||||||
else:
|
else:
|
||||||
try:
|
try:
|
||||||
result = mcp.record_activities(activities)
|
result = mcp.record_activities(activities)
|
||||||
mark_imported(tx_id, detail_hash, status, [str(i) for i in range(len(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)})
|
audit("imported", {"tx_id": tx_id, "activities": len(activities)})
|
||||||
stats["new"] += len(activities)
|
stats["new"] += len(activities)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
|||||||
Reference in New Issue
Block a user