323 lines
11 KiB
Python
323 lines
11 KiB
Python
"""
|
||
saxo_broker.py — Saxo Bank SIM paper trading integration.
|
||
|
||
Reads SAXO_TOKEN + SAXO_BASE from .env (or environment).
|
||
Token expires after 24h — refresh at developer.saxobank.com/openapi/token
|
||
"""
|
||
import os
|
||
import sys
|
||
import json
|
||
import time
|
||
import requests
|
||
from pathlib import Path
|
||
from datetime import datetime, timezone
|
||
from dotenv import load_dotenv
|
||
from db import get_conn
|
||
|
||
load_dotenv()
|
||
|
||
BASE = os.getenv("SAXO_BASE", "https://gateway.saxobank.com/sim/openapi")
|
||
|
||
def _get_token() -> str:
|
||
"""Get valid token — prefers OAuth2 stored token, falls back to 24h env token."""
|
||
try:
|
||
from saxo_auth import get_token
|
||
return get_token()
|
||
except Exception:
|
||
return os.getenv("SAXO_TOKEN", "")
|
||
|
||
# C25 tickers: our local ticker → (search keyword, Saxo ExchangeId)
|
||
# ExchangeId "CSE" = Copenhagen Stock Exchange (Saxo's code)
|
||
SAXO_SYMBOLS = {
|
||
"NOVO-B": ("Novo Nordisk B", "CSE"),
|
||
"DANSKE": ("Danske Bank", "CSE"),
|
||
"DSV": ("DSV", "CSE"),
|
||
"DEMANT": ("Demant", "CSE"),
|
||
"ORSTED": ("Orsted", "CSE"),
|
||
"ROCK-B": ("Rockwool", "CSE"),
|
||
"VWS": ("Vestas", "CSE"),
|
||
"COLOB": ("Coloplast", "CSE"),
|
||
"GMAB": ("Genmab", "CSE"),
|
||
"MAERSK-B": ("Maersk B", "CSE"),
|
||
"NZYM-B": ("Novozymes", "CSE"),
|
||
"NKT": ("NKT", "CSE"),
|
||
"GN": ("GN Store Nord", "CSE"),
|
||
"BAVAR": ("Bavarian Nordic", "CSE"),
|
||
"TRYG": ("Tryg", "CSE"),
|
||
"PNDORA": ("Pandora", "CSE"),
|
||
"CARL-B": ("Carlsberg", "CSE"),
|
||
"CHR": ("Chr. Hansen", "CSE"),
|
||
"FLS": ("FLSmidth", "CSE"),
|
||
"RBREW": ("Royal Unibrew", "CSE"),
|
||
"SYDB": ("Sydbank", "CSE"),
|
||
}
|
||
|
||
|
||
def _headers():
|
||
return {"Authorization": f"Bearer {_get_token()}", "Content-Type": "application/json"}
|
||
|
||
|
||
def _get(path, params=None):
|
||
r = requests.get(f"{BASE}{path}", headers=_headers(), params=params, timeout=10)
|
||
r.raise_for_status()
|
||
return r.json()
|
||
|
||
|
||
def _post(path, body):
|
||
r = requests.post(f"{BASE}{path}", headers=_headers(), json=body, timeout=10)
|
||
r.raise_for_status()
|
||
return r.json()
|
||
|
||
|
||
# ── Account info ────────────────────────────────────────────────────────────
|
||
|
||
def get_account_info():
|
||
"""Return (ClientKey, AccountKey) for the SIM account."""
|
||
data = _get("/port/v1/accounts/me")
|
||
accounts = data.get("Data", [])
|
||
if not accounts:
|
||
raise RuntimeError("No accounts found — are you logged into SIM?")
|
||
acct = accounts[0]
|
||
return acct["ClientKey"], acct["AccountKey"]
|
||
|
||
|
||
def get_balances():
|
||
"""Return cash balance dict from Saxo SIM."""
|
||
return _get("/port/v1/balances/me")
|
||
|
||
|
||
# ── Instrument lookup ───────────────────────────────────────────────────────
|
||
|
||
def find_uic(ticker: str) -> int | None:
|
||
"""Resolve our local ticker to Saxo UIC (instrument ID)."""
|
||
if ticker not in SAXO_SYMBOLS:
|
||
return None
|
||
symbol, exchange = SAXO_SYMBOLS[ticker]
|
||
results = _get("/ref/v1/instruments", params={
|
||
"Keywords": symbol,
|
||
"ExchangeId": exchange,
|
||
"AssetTypes": "Stock",
|
||
"$top": 5,
|
||
})
|
||
for item in results.get("Data", []):
|
||
if item.get("ExchangeId") == exchange:
|
||
return item["Identifier"]
|
||
return None
|
||
|
||
|
||
def get_price(uic: int) -> float | None:
|
||
"""
|
||
Get last traded price for a UIC.
|
||
NOTE: SIM 24h tokens have no market data access — returns None.
|
||
Use yfinance for prices instead (see signals.py).
|
||
"""
|
||
try:
|
||
data = _get("/trade/v1/infoprices", params={
|
||
"Uic": uic,
|
||
"AssetType": "Stock",
|
||
"FieldGroups": "Quote",
|
||
})
|
||
last = data.get("Quote", {}).get("Last")
|
||
return float(last) if last else None
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
# ── Orders ──────────────────────────────────────────────────────────────────
|
||
|
||
def place_order(
|
||
ticker: str,
|
||
shares: int,
|
||
direction: str = "Buy",
|
||
*,
|
||
price_dkk: float | None = None,
|
||
signal_score: float | None = None,
|
||
analyst_rec: str | None = None,
|
||
note: str | None = None,
|
||
) -> dict:
|
||
"""
|
||
Place a market order on Saxo SIM.
|
||
|
||
Args:
|
||
ticker: Our C25 ticker (e.g. "NOVO-B")
|
||
shares: Number of shares
|
||
direction: "Buy" or "Sell"
|
||
price_dkk: Estimated fill price in DKK (for fee calc + logging)
|
||
signal_score: NLP signal score that triggered this order
|
||
analyst_rec: Analyst recommendation label
|
||
note: Free-text note
|
||
|
||
Returns:
|
||
Saxo order response dict
|
||
"""
|
||
_, acct_key = get_account_info()
|
||
uic = find_uic(ticker)
|
||
if uic is None:
|
||
raise ValueError(f"Unknown ticker: {ticker}")
|
||
|
||
body = {
|
||
"Uic": uic,
|
||
"AssetType": "Stock",
|
||
"Amount": shares,
|
||
"BuySell": direction,
|
||
"OrderType": "Market",
|
||
"OrderDuration": {"DurationType": "DayOrder"},
|
||
"AccountKey": acct_key,
|
||
"ManualOrder": True,
|
||
}
|
||
resp = _post("/trade/v2/orders", body)
|
||
|
||
# Log to saxo_orders table regardless of outcome
|
||
saxo_order_id = resp.get("OrderId") or resp.get("Orders", [{}])[0].get("OrderId")
|
||
status = "filled" if saxo_order_id else "rejected"
|
||
log_order(
|
||
ticker=ticker,
|
||
direction=direction.lower(),
|
||
shares=shares,
|
||
price_dkk=price_dkk,
|
||
saxo_order_id=str(saxo_order_id) if saxo_order_id else None,
|
||
status=status,
|
||
signal_score=signal_score,
|
||
analyst_rec=analyst_rec,
|
||
note=note,
|
||
)
|
||
return resp
|
||
|
||
|
||
def log_order(
|
||
ticker: str,
|
||
direction: str,
|
||
shares: float,
|
||
price_dkk: float | None,
|
||
saxo_order_id: str | None = None,
|
||
status: str = "pending",
|
||
signal_score: float | None = None,
|
||
analyst_rec: str | None = None,
|
||
note: str | None = None,
|
||
) -> None:
|
||
"""
|
||
Write a trade record to saxo_orders table.
|
||
|
||
Fee model: max(25 DKK, total_dkk * 0.001)
|
||
"""
|
||
total_dkk = (shares * price_dkk) if price_dkk else None
|
||
fee_dkk = round(max(25.0, total_dkk * 0.001), 2) if total_dkk else None
|
||
now_ts = datetime.now(timezone.utc).isoformat()
|
||
|
||
db = get_conn()
|
||
db.execute(
|
||
"""INSERT INTO saxo_orders
|
||
(ticker, direction, shares, price_dkk, total_dkk, fee_dkk,
|
||
saxo_order_id, status, signal_score, analyst_rec, placed_at, note)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
|
||
(ticker, direction, shares, price_dkk, total_dkk, fee_dkk,
|
||
saxo_order_id, status, signal_score, analyst_rec, now_ts, note),
|
||
)
|
||
db.commit()
|
||
db.close()
|
||
|
||
|
||
# ── Positions ───────────────────────────────────────────────────────────────
|
||
|
||
def get_positions() -> list[dict]:
|
||
"""Return open positions from Saxo SIM."""
|
||
data = _get("/port/v1/positions/me", params={"FieldGroups": "PositionBase,PositionView"})
|
||
return data.get("Data", [])
|
||
|
||
|
||
def print_positions():
|
||
"""Pretty-print open positions."""
|
||
positions = get_positions()
|
||
if not positions:
|
||
print("Ingen åbne positioner i Saxo SIM")
|
||
return
|
||
print(f"\n{'Ticker':<12} {'Antal':>6} {'Åbn kurs':>10} {'Akt kurs':>10} {'P/L':>10}")
|
||
print("-" * 52)
|
||
for p in positions:
|
||
base = p.get("PositionBase", {})
|
||
view = p.get("PositionView", {})
|
||
symbol = base.get("Symbol", "?")
|
||
amount = base.get("Amount", 0)
|
||
open_price = base.get("OpenPrice", 0)
|
||
current = view.get("CurrentPrice", 0)
|
||
pl = view.get("ProfitLossOnTrade", 0)
|
||
pl_str = f"+{pl:.0f}" if pl >= 0 else f"{pl:.0f}"
|
||
print(f"{symbol:<12} {amount:>6} {open_price:>10.2f} {current:>10.2f} {pl_str:>10}")
|
||
|
||
|
||
def print_balances():
|
||
"""Pretty-print account balances."""
|
||
b = get_balances()
|
||
cash = b.get("CashBalance", 0)
|
||
equity = b.get("TotalValue", 0)
|
||
margin = b.get("MarginAvailableForTrading", 0)
|
||
currency = b.get("Currency", "EUR")
|
||
print(f"\n Kontant saldo: {cash:>12,.0f} {currency}")
|
||
print(f" Total portefølje: {equity:>12,.0f} {currency}")
|
||
print(f" Fri margin: {margin:>12,.0f} {currency}")
|
||
|
||
|
||
# ── CLI ──────────────────────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
try:
|
||
_get_token()
|
||
except Exception as e:
|
||
print(f"FEJL: {e}")
|
||
print("Kør: python saxo_auth.py login")
|
||
sys.exit(1)
|
||
|
||
cmd = sys.argv[1] if len(sys.argv) > 1 else "status"
|
||
|
||
if cmd == "status":
|
||
print("\n=== SAXO SIM KONTO ===")
|
||
client_key, acct_key = get_account_info()
|
||
print(f" ClientKey: {client_key}")
|
||
print(f" AccountKey: {acct_key}")
|
||
print_balances()
|
||
print_positions()
|
||
|
||
elif cmd == "buy":
|
||
# saxo_broker.py buy NOVO-B 5
|
||
if len(sys.argv) < 4:
|
||
print("Brug: saxo_broker.py buy TICKER ANTAL")
|
||
sys.exit(1)
|
||
ticker = sys.argv[2].upper()
|
||
shares = int(sys.argv[3])
|
||
print(f"Placerer KØB {shares} × {ticker} på Saxo SIM ...")
|
||
result = place_order(ticker, shares, "Buy")
|
||
print(f"Ordre bekræftet: {json.dumps(result, indent=2)}")
|
||
|
||
elif cmd == "sell":
|
||
# saxo_broker.py sell NOVO-B 5
|
||
if len(sys.argv) < 4:
|
||
print("Brug: saxo_broker.py sell TICKER ANTAL")
|
||
sys.exit(1)
|
||
ticker = sys.argv[2].upper()
|
||
shares = int(sys.argv[3])
|
||
print(f"Placerer SÆLG {shares} × {ticker} på Saxo SIM ...")
|
||
result = place_order(ticker, shares, "Sell")
|
||
print(f"Ordre bekræftet: {json.dumps(result, indent=2)}")
|
||
|
||
elif cmd == "positions":
|
||
print_positions()
|
||
|
||
elif cmd == "balances":
|
||
print_balances()
|
||
|
||
elif cmd == "uic":
|
||
# saxo_broker.py uic NOVO-B
|
||
ticker = sys.argv[2].upper() if len(sys.argv) > 2 else "NOVO-B"
|
||
uic = find_uic(ticker)
|
||
price = get_price(uic) if uic else None
|
||
print(f"{ticker}: UIC={uic}, pris={price}")
|
||
|
||
else:
|
||
print(f"Ukendt kommando: {cmd}")
|
||
print("Kommandoer: status | buy TICKER N | sell TICKER N | positions | balances | uic TICKER")
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|