provider: alpaca_omnisub — Alpaca Broker OmniSub (omnibus + sub-accounts)

Sibling to alpaca provider. Runs in parallel with standalone Alpaca using
independent ALPACA_OMNISUB_* env vars. Full Provider interface plus
per-sub positions, omnibus-routed transfers, batch journals + reverse-batch
for trade reversals, omnibus snapshot for reconciliation. 19 tests.
This commit is contained in:
2026-04-13 13:49:03 -07:00
parent f8d7400eac
commit 86afa4ba31
10 changed files with 2207 additions and 1 deletions
+27
View File
@@ -0,0 +1,27 @@
# alpaca_omnisub
Alpaca OmniSub (Omnibus + Sub-accounts) provider for lux/broker.
## How it differs from `alpaca`
| | `alpaca` | `alpaca_omnisub` |
|---|---|---|
| Account model | Standalone individual/joint accounts | Sub-accounts under a single omnibus master |
| Account creation | `account_type` = individual/joint | `account_type` = `omnibus_sub_account`, `omnibus_master_id` set |
| Orders | Scoped to each account | Scoped to each sub-account |
| Positions | Per-account | Per-sub-account |
| Transfers (ACH/wire) | Per-account | Routed through omnibus master; internal ledger via journals |
| Bank relationships | Per-account | On omnibus master |
| Reconciliation | Per-account | Omnibus aggregate via `GetOmnibusSnapshot()` |
| Market data | Alpaca data API | Same Alpaca data API |
## Environment variables
```
ALPACA_OMNISUB_API_KEY # Alpaca broker API key (omnisub-specific)
ALPACA_OMNISUB_API_SECRET # Alpaca broker API secret
ALPACA_OMNISUB_BASE_URL # defaults to sandbox
ALPACA_OMNISUB_OMNIBUS_ACCOUNT_ID # the omnibus master account ID
```
Both `alpaca` and `alpaca_omnisub` can run in parallel on the same registry with independent credentials.
@@ -0,0 +1,595 @@
// Package alpaca_omnisub implements the broker Provider interface for
// Alpaca's OmniSub (Omnibus + Sub-accounts) model. Sub-accounts are
// created under a single omnibus master; orders and positions are
// scoped per-sub, while cash flows through the omnibus aggregate.
package alpaca_omnisub
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"strconv"
"strings"
"time"
"github.com/luxfi/broker/pkg/types"
)
const (
SandboxURL = "https://broker-api.sandbox.alpaca.markets"
ProductionURL = "https://broker-api.alpaca.markets"
DataURL = "https://data.alpaca.markets"
DataSandboxURL = "https://data.sandbox.alpaca.markets"
)
// Config for the Alpaca OmniSub provider.
type Config struct {
BaseURL string `json:"base_url"`
APIKey string `json:"-"`
APISecret string `json:"-"`
OmnibusAccountID string `json:"omnibus_account_id"`
}
// Provider implements the broker Provider interface for Alpaca OmniSub.
type Provider struct {
cfg Config
client *http.Client
dataURL string
}
func New(cfg Config) *Provider {
if cfg.BaseURL == "" {
cfg.BaseURL = SandboxURL
}
dataURL := DataURL
if strings.Contains(cfg.BaseURL, "sandbox") {
dataURL = DataSandboxURL
}
return &Provider{
cfg: cfg,
dataURL: dataURL,
client: &http.Client{
Timeout: 30 * time.Second,
},
}
}
func (p *Provider) Name() string { return "alpaca_omnisub" }
// --- HTTP helpers ---
func (p *Provider) do(ctx context.Context, method, path string, body interface{}) ([]byte, int, error) {
var reqBody io.Reader
if body != nil {
b, err := json.Marshal(body)
if err != nil {
return nil, 0, fmt.Errorf("marshal: %w", err)
}
reqBody = bytes.NewReader(b)
}
fullURL := p.cfg.BaseURL + path
req, err := http.NewRequestWithContext(ctx, method, fullURL, reqBody)
if err != nil {
return nil, 0, err
}
req.SetBasicAuth(p.cfg.APIKey, p.cfg.APISecret)
req.Header.Set("Content-Type", "application/json")
resp, err := p.client.Do(req)
if err != nil {
return nil, 0, err
}
defer resp.Body.Close()
data, err := io.ReadAll(resp.Body)
if err != nil {
return nil, resp.StatusCode, err
}
if resp.StatusCode >= 400 {
var apiErr struct {
Code int `json:"code"`
Message string `json:"message"`
}
if json.Unmarshal(data, &apiErr) == nil && apiErr.Message != "" {
return nil, resp.StatusCode, fmt.Errorf("alpaca_omnisub %d: %s", apiErr.Code, apiErr.Message)
}
return nil, resp.StatusCode, fmt.Errorf("alpaca_omnisub %d: %s", resp.StatusCode, string(data))
}
return data, resp.StatusCode, nil
}
func (p *Provider) doData(ctx context.Context, method, path string) ([]byte, int, error) {
req, err := http.NewRequestWithContext(ctx, method, p.dataURL+path, nil)
if err != nil {
return nil, 0, err
}
req.SetBasicAuth(p.cfg.APIKey, p.cfg.APISecret)
req.Header.Set("Content-Type", "application/json")
resp, err := p.client.Do(req)
if err != nil {
return nil, 0, err
}
defer resp.Body.Close()
data, err := io.ReadAll(resp.Body)
if err != nil {
return nil, resp.StatusCode, err
}
if resp.StatusCode >= 400 {
return nil, resp.StatusCode, fmt.Errorf("alpaca_omnisub data %d: %s", resp.StatusCode, string(data))
}
return data, resp.StatusCode, nil
}
// --- Market Data (shared Alpaca data API, identical to standalone) ---
func isCryptoSymbol(symbol string) bool {
return strings.Contains(symbol, "/")
}
func stocksOrCryptoPath(symbol string) string {
if isCryptoSymbol(symbol) {
return "/v1beta3/crypto/us"
}
return "/v2/stocks"
}
func (p *Provider) GetSnapshot(ctx context.Context, symbol string) (*types.MarketSnapshot, error) {
if isCryptoSymbol(symbol) {
path := "/v1beta3/crypto/us/snapshots?symbols=" + url.QueryEscape(symbol)
data, _, err := p.doData(ctx, http.MethodGet, path)
if err != nil {
return nil, err
}
var wrapper struct {
Snapshots map[string]json.RawMessage `json:"snapshots"`
}
if err := json.Unmarshal(data, &wrapper); err != nil {
return nil, fmt.Errorf("parse crypto snapshots: %w", err)
}
raw, ok := wrapper.Snapshots[symbol]
if !ok {
return nil, fmt.Errorf("crypto snapshot not found for %s", symbol)
}
return p.parseSnapshot(symbol, raw)
}
data, _, err := p.doData(ctx, http.MethodGet, "/v2/stocks/"+symbol+"/snapshot")
if err != nil {
return nil, err
}
return p.parseSnapshot(symbol, data)
}
func (p *Provider) GetSnapshots(ctx context.Context, symbols []string) (map[string]*types.MarketSnapshot, error) {
var stocks, cryptos []string
for _, s := range symbols {
if isCryptoSymbol(s) {
cryptos = append(cryptos, s)
} else {
stocks = append(stocks, s)
}
}
result := make(map[string]*types.MarketSnapshot)
if len(stocks) > 0 {
path := "/v2/stocks/snapshots?symbols=" + url.QueryEscape(strings.Join(stocks, ","))
data, _, err := p.doData(ctx, http.MethodGet, path)
if err != nil {
return nil, err
}
var raw map[string]json.RawMessage
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
for sym, d := range raw {
snap, err := p.parseSnapshot(sym, d)
if err != nil {
continue
}
result[sym] = snap
}
}
if len(cryptos) > 0 {
path := "/v1beta3/crypto/us/snapshots?symbols=" + url.QueryEscape(strings.Join(cryptos, ","))
data, _, err := p.doData(ctx, http.MethodGet, path)
if err != nil {
return nil, err
}
var wrapper struct {
Snapshots map[string]json.RawMessage `json:"snapshots"`
}
if err := json.Unmarshal(data, &wrapper); err != nil {
return nil, err
}
for sym, d := range wrapper.Snapshots {
snap, err := p.parseSnapshot(sym, d)
if err != nil {
continue
}
result[sym] = snap
}
}
return result, nil
}
func (p *Provider) parseSnapshot(symbol string, data []byte) (*types.MarketSnapshot, error) {
var raw struct {
LatestTrade *struct {
T string `json:"t"`
P float64 `json:"p"`
S float64 `json:"s"`
X string `json:"x"`
} `json:"latestTrade"`
LatestQuote *struct {
T string `json:"t"`
BP float64 `json:"bp"`
BS float64 `json:"bs"`
AP float64 `json:"ap"`
AS float64 `json:"as"`
} `json:"latestQuote"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
snap := &types.MarketSnapshot{Symbol: symbol}
if raw.LatestTrade != nil {
snap.LatestTrade = &types.Trade{Timestamp: raw.LatestTrade.T, Price: raw.LatestTrade.P, Size: raw.LatestTrade.S, Exchange: raw.LatestTrade.X}
}
if raw.LatestQuote != nil {
snap.LatestQuote = &types.Quote{Timestamp: raw.LatestQuote.T, BidPrice: raw.LatestQuote.BP, BidSize: raw.LatestQuote.BS, AskPrice: raw.LatestQuote.AP, AskSize: raw.LatestQuote.AS}
}
return snap, nil
}
func (p *Provider) GetBars(ctx context.Context, symbol, timeframe, start, end string, limit int) ([]*types.Bar, error) {
var allBars []*types.Bar
nextPageToken := ""
crypto := isCryptoSymbol(symbol)
for page := 0; page < 100; page++ {
var path string
if crypto {
path = "/v1beta3/crypto/us/bars?symbols=" + symbol + "&timeframe=" + timeframe
} else {
path = stocksOrCryptoPath(symbol) + "/" + symbol + "/bars?timeframe=" + timeframe
}
if start != "" {
path += "&start=" + start
}
if end != "" {
path += "&end=" + end
}
if limit > 0 {
path += "&limit=" + strconv.Itoa(limit)
}
if nextPageToken != "" {
path += "&page_token=" + nextPageToken
}
data, _, err := p.doData(ctx, http.MethodGet, path)
if err != nil {
return nil, err
}
type barEntry struct {
T string `json:"t"`
O float64 `json:"o"`
H float64 `json:"h"`
L float64 `json:"l"`
C float64 `json:"c"`
V float64 `json:"v"`
VW float64 `json:"vw"`
N int `json:"n"`
}
var bars []barEntry
var nextToken string
if crypto {
var raw struct {
Bars map[string][]barEntry `json:"bars"`
NextPageToken string `json:"next_page_token"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
bars = raw.Bars[symbol]
nextToken = raw.NextPageToken
} else {
var raw struct {
Bars []barEntry `json:"bars"`
NextPageToken string `json:"next_page_token"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
bars = raw.Bars
nextToken = raw.NextPageToken
}
for _, b := range bars {
var timeMs int64
if t, err := time.Parse(time.RFC3339, b.T); err == nil {
timeMs = t.UnixMilli()
} else if t, err := time.Parse(time.RFC3339Nano, b.T); err == nil {
timeMs = t.UnixMilli()
}
allBars = append(allBars, &types.Bar{
Timestamp: b.T,
TimeMs: timeMs,
Open: b.O,
High: b.H,
Low: b.L,
Close: b.C,
Volume: b.V,
VWAP: b.VW,
TradeCount: b.N,
})
}
if nextToken == "" {
break
}
if limit > 0 && len(allBars) >= limit {
allBars = allBars[:limit]
break
}
nextPageToken = nextToken
}
return allBars, nil
}
func (p *Provider) GetLatestTrades(ctx context.Context, symbols []string) (map[string]*types.Trade, error) {
result := make(map[string]*types.Trade)
var stocks, cryptos []string
for _, s := range symbols {
if isCryptoSymbol(s) {
cryptos = append(cryptos, s)
} else {
stocks = append(stocks, s)
}
}
if len(stocks) > 0 {
path := "/v2/stocks/trades/latest?symbols=" + strings.Join(stocks, ",")
data, _, err := p.doData(ctx, http.MethodGet, path)
if err != nil {
return nil, err
}
var raw struct {
Trades map[string]struct {
T string `json:"t"`
P float64 `json:"p"`
S float64 `json:"s"`
X string `json:"x"`
} `json:"trades"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
for sym, t := range raw.Trades {
result[sym] = &types.Trade{Timestamp: t.T, Price: t.P, Size: t.S, Exchange: t.X}
}
}
if len(cryptos) > 0 {
path := "/v1beta3/crypto/us/latest/trades?symbols=" + strings.Join(cryptos, ",")
data, _, err := p.doData(ctx, http.MethodGet, path)
if err != nil {
return nil, err
}
var raw struct {
Trades map[string]struct {
T string `json:"t"`
P float64 `json:"p"`
S float64 `json:"s"`
} `json:"trades"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
for sym, t := range raw.Trades {
result[sym] = &types.Trade{Timestamp: t.T, Price: t.P, Size: t.S}
}
}
return result, nil
}
func (p *Provider) GetLatestQuotes(ctx context.Context, symbols []string) (map[string]*types.Quote, error) {
result := make(map[string]*types.Quote)
var stocks, cryptos []string
for _, s := range symbols {
if isCryptoSymbol(s) {
cryptos = append(cryptos, s)
} else {
stocks = append(stocks, s)
}
}
if len(stocks) > 0 {
path := "/v2/stocks/quotes/latest?symbols=" + strings.Join(stocks, ",")
data, _, err := p.doData(ctx, http.MethodGet, path)
if err != nil {
return nil, err
}
var raw struct {
Quotes map[string]struct {
T string `json:"t"`
BP float64 `json:"bp"`
BS float64 `json:"bs"`
AP float64 `json:"ap"`
AS float64 `json:"as"`
} `json:"quotes"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
for sym, q := range raw.Quotes {
result[sym] = &types.Quote{Timestamp: q.T, BidPrice: q.BP, BidSize: q.BS, AskPrice: q.AP, AskSize: q.AS}
}
}
if len(cryptos) > 0 {
path := "/v1beta3/crypto/us/latest/quotes?symbols=" + strings.Join(cryptos, ",")
data, _, err := p.doData(ctx, http.MethodGet, path)
if err != nil {
return nil, err
}
var raw struct {
Quotes map[string]struct {
T string `json:"t"`
BP float64 `json:"bp"`
BS float64 `json:"bs"`
AP float64 `json:"ap"`
AS float64 `json:"as"`
} `json:"quotes"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
for sym, q := range raw.Quotes {
result[sym] = &types.Quote{Timestamp: q.T, BidPrice: q.BP, BidSize: q.BS, AskPrice: q.AP, AskSize: q.AS}
}
}
return result, nil
}
func (p *Provider) GetClock(ctx context.Context) (*types.MarketClock, error) {
data, _, err := p.do(ctx, http.MethodGet, "/v1/clock", nil)
if err != nil {
return nil, err
}
var raw struct {
Timestamp string `json:"timestamp"`
IsOpen bool `json:"is_open"`
NextOpen string `json:"next_open"`
NextClose string `json:"next_close"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
return &types.MarketClock{
Timestamp: raw.Timestamp,
IsOpen: raw.IsOpen,
NextOpen: raw.NextOpen,
NextClose: raw.NextClose,
}, nil
}
func (p *Provider) GetCalendar(ctx context.Context, start, end string) ([]*types.MarketCalendarDay, error) {
path := "/v1/calendar"
sep := "?"
if start != "" {
path += sep + "start=" + start
sep = "&"
}
if end != "" {
path += sep + "end=" + end
}
data, _, err := p.do(ctx, http.MethodGet, path, nil)
if err != nil {
return nil, err
}
var raw []struct {
Date string `json:"date"`
Open string `json:"open"`
Close string `json:"close"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
days := make([]*types.MarketCalendarDay, len(raw))
for i, d := range raw {
days[i] = &types.MarketCalendarDay{Date: d.Date, Open: d.Open, Close: d.Close}
}
return days, nil
}
// --- Assets (same Alpaca API, different provider tag) ---
func (p *Provider) ListAssets(ctx context.Context, class string) ([]*types.Asset, error) {
path := "/v1/assets?status=active"
if class != "" {
path += "&asset_class=" + class
}
data, _, err := p.do(ctx, http.MethodGet, path, nil)
if err != nil {
return nil, err
}
var raw []struct {
ID string `json:"id"`
Symbol string `json:"symbol"`
Name string `json:"name"`
Class string `json:"class"`
Exchange string `json:"exchange"`
Status string `json:"status"`
Tradable bool `json:"tradable"`
Fractionable bool `json:"fractionable"`
MinOrderSize string `json:"min_order_size"`
PriceIncrement string `json:"price_increment"`
MinTradeIncrement string `json:"min_trade_increment"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
assets := make([]*types.Asset, len(raw))
for i, r := range raw {
assets[i] = &types.Asset{
ID: r.ID,
Provider: "alpaca_omnisub",
Symbol: r.Symbol,
Name: r.Name,
Class: r.Class,
Exchange: r.Exchange,
Status: r.Status,
Tradable: r.Tradable,
Fractionable: r.Fractionable,
MinOrderSize: r.MinOrderSize,
PriceIncrement: r.PriceIncrement,
MinTradeIncrement: r.MinTradeIncrement,
}
}
return assets, nil
}
func (p *Provider) GetAsset(ctx context.Context, symbolOrID string) (*types.Asset, error) {
apiSymbol := strings.ReplaceAll(symbolOrID, "/", "")
data, _, err := p.do(ctx, http.MethodGet, "/v1/assets/"+apiSymbol, nil)
if err != nil {
return nil, err
}
var raw struct {
ID string `json:"id"`
Symbol string `json:"symbol"`
Name string `json:"name"`
Class string `json:"class"`
Exchange string `json:"exchange"`
Status string `json:"status"`
Tradable bool `json:"tradable"`
Fractionable bool `json:"fractionable"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
return &types.Asset{
ID: raw.ID,
Provider: "alpaca_omnisub",
Symbol: raw.Symbol,
Name: raw.Name,
Class: raw.Class,
Exchange: raw.Exchange,
Status: raw.Status,
Tradable: raw.Tradable,
Fractionable: raw.Fractionable,
}, nil
}
@@ -0,0 +1,194 @@
package alpaca_omnisub
import (
"context"
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/luxfi/broker/pkg/types"
)
type alpacaAccount struct {
ID string `json:"id"`
AccountNumber string `json:"account_number"`
Status string `json:"status"`
Currency string `json:"currency"`
AccountType string `json:"account_type"`
CreatedAt string `json:"created_at"`
Contact *struct {
EmailAddress string `json:"email_address"`
PhoneNumber string `json:"phone_number"`
StreetAddress []string `json:"street_address"`
City string `json:"city"`
State string `json:"state"`
PostalCode string `json:"postal_code"`
Country string `json:"country"`
} `json:"contact"`
Identity *struct {
GivenName string `json:"given_name"`
FamilyName string `json:"family_name"`
DateOfBirth string `json:"date_of_birth"`
TaxIDType string `json:"tax_id_type"`
CountryOfTaxResidence string `json:"country_of_tax_residence"`
} `json:"identity"`
EnabledAssets []string `json:"enabled_assets"`
}
func (a *alpacaAccount) toUnified() *types.Account {
acct := &types.Account{
ID: a.ID,
Provider: "alpaca_omnisub",
ProviderID: a.ID,
AccountNumber: a.AccountNumber,
Status: a.Status,
Currency: a.Currency,
AccountType: a.AccountType,
EnabledAssets: a.EnabledAssets,
}
if t, err := time.Parse(time.RFC3339Nano, a.CreatedAt); err == nil {
acct.CreatedAt = t
}
if a.Identity != nil {
acct.Identity = &types.Identity{
GivenName: a.Identity.GivenName,
FamilyName: a.Identity.FamilyName,
DateOfBirth: a.Identity.DateOfBirth,
TaxIDType: a.Identity.TaxIDType,
CountryOfTax: a.Identity.CountryOfTaxResidence,
}
}
if a.Contact != nil {
acct.Contact = &types.Contact{
Email: a.Contact.EmailAddress,
Phone: a.Contact.PhoneNumber,
Street: a.Contact.StreetAddress,
City: a.Contact.City,
State: a.Contact.State,
PostalCode: a.Contact.PostalCode,
Country: a.Contact.Country,
}
}
return acct
}
// CreateAccount creates an omnibus sub-account under the configured master.
// The Alpaca API requires account_type=omnibus_sub_account and the
// omnibus_master_id field referencing the omnibus master account.
func (p *Provider) CreateAccount(ctx context.Context, req *types.CreateAccountRequest) (*types.Account, error) {
now := time.Now().UTC().Format(time.RFC3339)
ipAddress := req.IPAddress
if ipAddress == "" {
return nil, errMissingIP
}
body := map[string]interface{}{
"account_type": "omnibus_sub_account",
"omnibus_master_id": p.cfg.OmnibusAccountID,
"contact": map[string]interface{}{
"email_address": req.Contact.Email,
"phone_number": req.Contact.Phone,
"street_address": req.Contact.Street,
"city": req.Contact.City,
"state": req.Contact.State,
"postal_code": req.Contact.PostalCode,
"country": req.Contact.Country,
},
"identity": map[string]interface{}{
"given_name": req.Identity.GivenName,
"family_name": req.Identity.FamilyName,
"date_of_birth": req.Identity.DateOfBirth,
"tax_id": req.Identity.TaxID,
"tax_id_type": req.Identity.TaxIDType,
"country_of_tax_residence": req.Identity.CountryOfTax,
"funding_source": fundingSources(req.FundingSources),
},
"disclosures": disclosuresBody(req.Disclosures),
"agreements": []map[string]interface{}{
{"agreement": "margin_agreement", "signed_at": now, "ip_address": ipAddress},
{"agreement": "account_agreement", "signed_at": now, "ip_address": ipAddress},
{"agreement": "customer_agreement", "signed_at": now, "ip_address": ipAddress},
},
}
if len(req.EnabledAssets) > 0 {
body["enabled_assets"] = req.EnabledAssets
}
data, _, err := p.do(ctx, http.MethodPost, "/v1/accounts", body)
if err != nil {
return nil, err
}
var raw alpacaAccount
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
return raw.toUnified(), nil
}
// GetAccount retrieves a sub-account by its Alpaca ID.
func (p *Provider) GetAccount(ctx context.Context, providerAccountID string) (*types.Account, error) {
data, _, err := p.do(ctx, http.MethodGet, "/v1/accounts/"+providerAccountID, nil)
if err != nil {
return nil, err
}
var raw alpacaAccount
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
return raw.toUnified(), nil
}
// ListAccounts lists all sub-accounts under this omnibus master.
// Alpaca filters by query=omnibus_master_id.
func (p *Provider) ListAccounts(ctx context.Context) ([]*types.Account, error) {
path := "/v1/accounts?query=" + p.cfg.OmnibusAccountID
data, _, err := p.do(ctx, http.MethodGet, path, nil)
if err != nil {
return nil, err
}
var raw []alpacaAccount
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
accts := make([]*types.Account, len(raw))
for i := range raw {
accts[i] = raw[i].toUnified()
}
return accts, nil
}
// --- helpers ---
var errMissingIP = fmt.Errorf("client IP address is required for agreement signing")
func fundingSources(src []string) []string {
if len(src) == 0 {
return []string{"employment_income"}
}
return src
}
func disclosuresBody(d *types.AccountDisclosures) map[string]interface{} {
cp, af, pe, fe := false, false, false, false
if d != nil {
if d.IsControlPerson != nil {
cp = *d.IsControlPerson
}
if d.IsAffiliatedExchangeFinra != nil {
af = *d.IsAffiliatedExchangeFinra
}
if d.IsPoliticallyExposed != nil {
pe = *d.IsPoliticallyExposed
}
if d.ImmediateFamilyExposed != nil {
fe = *d.ImmediateFamilyExposed
}
}
return map[string]interface{}{
"is_control_person": cp,
"is_affiliated_exchange_or_finra": af,
"is_politically_exposed": pe,
"immediate_family_exposed": fe,
}
}
@@ -0,0 +1,205 @@
package alpaca_omnisub
import (
"context"
"encoding/json"
"fmt"
"net/http"
"github.com/luxfi/broker/pkg/types"
)
// CreateJournal creates an inter-sub-account journal entry.
// In OmniSub, journals move cash (JNLC) or securities (JNLS) between
// sub-accounts or between the omnibus master and a sub-account.
func (p *Provider) CreateJournal(ctx context.Context, req *types.CreateJournalRequest) (*types.Journal, error) {
body := map[string]interface{}{
"entry_type": req.EntryType,
"from_account": req.FromAccount,
"to_account": req.ToAccount,
}
if req.Amount != "" {
body["amount"] = req.Amount
}
if req.Symbol != "" {
body["symbol"] = req.Symbol
}
if req.Qty != "" {
body["qty"] = req.Qty
}
if req.Description != "" {
body["description"] = req.Description
}
data, _, err := p.do(ctx, http.MethodPost, "/v1/journals", body)
if err != nil {
return nil, err
}
return parseJournal(data)
}
// ListJournals lists journal entries, optionally filtered.
func (p *Provider) ListJournals(ctx context.Context, params *types.JournalParams) ([]*types.Journal, error) {
path := "/v1/journals"
sep := "?"
if params != nil {
if params.After != "" {
path += sep + "after=" + params.After
sep = "&"
}
if params.Before != "" {
path += sep + "before=" + params.Before
sep = "&"
}
if params.Status != "" {
path += sep + "status=" + params.Status
sep = "&"
}
if params.EntryType != "" {
path += sep + "entry_type=" + params.EntryType
sep = "&"
}
if params.ToAccount != "" {
path += sep + "to_account=" + params.ToAccount
sep = "&"
}
if params.FromAccount != "" {
path += sep + "from_account=" + params.FromAccount
}
}
data, _, err := p.do(ctx, http.MethodGet, path, nil)
if err != nil {
return nil, err
}
var raw []json.RawMessage
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
journals := make([]*types.Journal, 0, len(raw))
for _, r := range raw {
j, err := parseJournal(r)
if err != nil {
continue
}
journals = append(journals, j)
}
return journals, nil
}
// GetJournal retrieves a single journal entry by ID.
func (p *Provider) GetJournal(ctx context.Context, journalID string) (*types.Journal, error) {
data, _, err := p.do(ctx, http.MethodGet, "/v1/journals/"+journalID, nil)
if err != nil {
return nil, err
}
return parseJournal(data)
}
// DeleteJournal cancels a pending journal entry.
func (p *Provider) DeleteJournal(ctx context.Context, journalID string) error {
_, _, err := p.do(ctx, http.MethodDelete, "/v1/journals/"+journalID, nil)
return err
}
// CreateBatchJournal creates up to 5000 journal entries in one call.
func (p *Provider) CreateBatchJournal(ctx context.Context, req *types.BatchJournalRequest) ([]*types.Journal, error) {
if len(req.Entries) > 5000 {
return nil, fmt.Errorf("batch journal entries %d exceeds limit of 5000", len(req.Entries))
}
body := map[string]interface{}{
"entry_type": req.EntryType,
"from_account": req.FromAccount,
"entries": req.Entries,
}
if req.Description != "" {
body["description"] = req.Description
}
data, _, err := p.do(ctx, http.MethodPost, "/v1/journals/batch", body)
if err != nil {
return nil, err
}
var raw []json.RawMessage
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
journals := make([]*types.Journal, 0, len(raw))
for _, r := range raw {
j, err := parseJournal(r)
if err != nil {
continue
}
journals = append(journals, j)
}
return journals, nil
}
// ReverseBatchJournal reverses a batch of journal entries.
func (p *Provider) ReverseBatchJournal(ctx context.Context, req *types.ReverseBatchJournalRequest) ([]*types.Journal, error) {
body := map[string]interface{}{
"entry_type": req.EntryType,
"from_account": req.FromAccount,
"entries": req.Entries,
}
if req.Description != "" {
body["description"] = req.Description
}
data, _, err := p.do(ctx, http.MethodPost, "/v1/journals/reverse_batch", body)
if err != nil {
return nil, err
}
var raw []json.RawMessage
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
journals := make([]*types.Journal, 0, len(raw))
for _, r := range raw {
j, err := parseJournal(r)
if err != nil {
continue
}
journals = append(journals, j)
}
return journals, nil
}
func parseJournal(data []byte) (*types.Journal, error) {
var raw struct {
ID string `json:"id"`
EntryType string `json:"entry_type"`
FromAccount string `json:"from_account"`
ToAccount string `json:"to_account"`
Symbol string `json:"symbol"`
Qty string `json:"qty"`
Price string `json:"price"`
Amount string `json:"net_amount"`
Status string `json:"status"`
Description string `json:"description"`
SettleDate string `json:"settle_date"`
SystemDate string `json:"system_date"`
CreatedAt string `json:"created_at"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
return &types.Journal{
ID: raw.ID,
EntryType: raw.EntryType,
FromAccount: raw.FromAccount,
ToAccount: raw.ToAccount,
Symbol: raw.Symbol,
Qty: raw.Qty,
Price: raw.Price,
Amount: raw.Amount,
Status: raw.Status,
Description: raw.Description,
SettleDate: raw.SettleDate,
SystemDate: raw.SystemDate,
CreatedAt: raw.CreatedAt,
}, nil
}
@@ -0,0 +1,154 @@
package alpaca_omnisub
import (
"context"
"encoding/json"
"net/http"
"time"
"github.com/luxfi/broker/pkg/types"
)
// CreateOrder places an order scoped to a sub-account.
// Alpaca OmniSub routes: POST /v1/trading/accounts/{subID}/orders
func (p *Provider) CreateOrder(ctx context.Context, subAccountID string, req *types.CreateOrderRequest) (*types.Order, error) {
body := map[string]interface{}{
"symbol": req.Symbol,
"side": req.Side,
"type": req.Type,
"time_in_force": req.TimeInForce,
}
if req.Qty != "" {
body["qty"] = req.Qty
}
if req.Notional != "" {
body["notional"] = req.Notional
}
if req.LimitPrice != "" {
body["limit_price"] = req.LimitPrice
}
if req.StopPrice != "" {
body["stop_price"] = req.StopPrice
}
if req.ClientOrderID != "" {
body["client_order_id"] = req.ClientOrderID
}
if req.TrailPrice != "" {
body["trail_price"] = req.TrailPrice
}
if req.TrailPercent != "" {
body["trail_percent"] = req.TrailPercent
}
if req.ExtendedHours {
body["extended_hours"] = true
}
if req.OrderClass != "" {
body["order_class"] = req.OrderClass
}
if req.TakeProfit != nil {
body["take_profit"] = map[string]interface{}{
"limit_price": req.TakeProfit.LimitPrice,
}
}
if req.StopLoss != nil {
sl := map[string]interface{}{
"stop_price": req.StopLoss.StopPrice,
}
if req.StopLoss.LimitPrice != "" {
sl["limit_price"] = req.StopLoss.LimitPrice
}
body["stop_loss"] = sl
}
data, _, err := p.do(ctx, http.MethodPost, "/v1/trading/accounts/"+subAccountID+"/orders", body)
if err != nil {
return nil, err
}
return p.parseOrder(data)
}
// ListOrders lists all orders for a sub-account.
func (p *Provider) ListOrders(ctx context.Context, subAccountID string) ([]*types.Order, error) {
path := "/v1/trading/accounts/" + subAccountID + "/orders?status=all"
data, _, err := p.do(ctx, http.MethodGet, path, nil)
if err != nil {
return nil, err
}
var raw []json.RawMessage
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
orders := make([]*types.Order, 0, len(raw))
for _, r := range raw {
o, err := p.parseOrder(r)
if err != nil {
continue
}
orders = append(orders, o)
}
return orders, nil
}
// GetOrder retrieves a single order for a sub-account.
func (p *Provider) GetOrder(ctx context.Context, subAccountID, orderID string) (*types.Order, error) {
data, _, err := p.do(ctx, http.MethodGet, "/v1/trading/accounts/"+subAccountID+"/orders/"+orderID, nil)
if err != nil {
return nil, err
}
return p.parseOrder(data)
}
// CancelOrder cancels an order for a sub-account.
func (p *Provider) CancelOrder(ctx context.Context, subAccountID, orderID string) error {
_, _, err := p.do(ctx, http.MethodDelete, "/v1/trading/accounts/"+subAccountID+"/orders/"+orderID, nil)
return err
}
func (p *Provider) parseOrder(data []byte) (*types.Order, error) {
var raw struct {
ID string `json:"id"`
Symbol string `json:"symbol"`
Qty string `json:"qty"`
Notional string `json:"notional"`
Side string `json:"side"`
Type string `json:"type"`
TimeInForce string `json:"time_in_force"`
LimitPrice string `json:"limit_price"`
StopPrice string `json:"stop_price"`
Status string `json:"status"`
FilledQty string `json:"filled_qty"`
FilledAvgPrice string `json:"filled_avg_price"`
AssetClass string `json:"asset_class"`
CreatedAt string `json:"created_at"`
FilledAt *string `json:"filled_at"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
o := &types.Order{
Provider: "alpaca_omnisub",
ProviderID: raw.ID,
Symbol: raw.Symbol,
Qty: raw.Qty,
Notional: raw.Notional,
Side: raw.Side,
Type: raw.Type,
TimeInForce: raw.TimeInForce,
LimitPrice: raw.LimitPrice,
StopPrice: raw.StopPrice,
Status: raw.Status,
FilledQty: raw.FilledQty,
FilledAvgPrice: raw.FilledAvgPrice,
AssetClass: raw.AssetClass,
}
if t, err := time.Parse(time.RFC3339Nano, raw.CreatedAt); err == nil {
o.CreatedAt = t
}
if raw.FilledAt != nil {
if t, err := time.Parse(time.RFC3339Nano, *raw.FilledAt); err == nil {
o.FilledAt = &t
}
}
return o, nil
}
@@ -0,0 +1,67 @@
package alpaca_omnisub
import (
"context"
"encoding/json"
"net/http"
"github.com/luxfi/broker/pkg/types"
)
// GetPortfolio returns account + positions for a sub-account.
func (p *Provider) GetPortfolio(ctx context.Context, subAccountID string) (*types.Portfolio, error) {
tData, _, err := p.do(ctx, http.MethodGet, "/v1/trading/accounts/"+subAccountID+"/account", nil)
if err != nil {
return nil, err
}
var ta struct {
Cash string `json:"cash"`
Equity string `json:"equity"`
BuyingPower string `json:"buying_power"`
PortfolioValue string `json:"portfolio_value"`
}
if err := json.Unmarshal(tData, &ta); err != nil {
return nil, err
}
pData, _, err := p.do(ctx, http.MethodGet, "/v1/trading/accounts/"+subAccountID+"/positions", nil)
if err != nil {
return nil, err
}
var rawPositions []struct {
Symbol string `json:"symbol"`
Qty string `json:"qty"`
AvgEntryPrice string `json:"avg_entry_price"`
MarketValue string `json:"market_value"`
CurrentPrice string `json:"current_price"`
UnrealizedPL string `json:"unrealized_pl"`
Side string `json:"side"`
AssetClass string `json:"asset_class"`
}
if err := json.Unmarshal(pData, &rawPositions); err != nil {
return nil, err
}
positions := make([]types.Position, len(rawPositions))
for i, rp := range rawPositions {
positions[i] = types.Position{
Symbol: rp.Symbol,
Qty: rp.Qty,
AvgEntryPrice: rp.AvgEntryPrice,
MarketValue: rp.MarketValue,
CurrentPrice: rp.CurrentPrice,
UnrealizedPL: rp.UnrealizedPL,
Side: rp.Side,
AssetClass: rp.AssetClass,
}
}
return &types.Portfolio{
AccountID: subAccountID,
Cash: ta.Cash,
Equity: ta.Equity,
BuyingPower: ta.BuyingPower,
PortfolioValue: ta.PortfolioValue,
Positions: positions,
}, nil
}
@@ -0,0 +1,81 @@
package alpaca_omnisub
import (
"context"
"encoding/json"
"net/http"
"github.com/luxfi/broker/pkg/types"
)
// OmnibusSnapshot is the end-of-day reconciliation snapshot for the
// omnibus master account. It includes the aggregate cash, equity,
// buying power, and all positions held at the omnibus level.
type OmnibusSnapshot struct {
AccountID string `json:"account_id"`
Cash string `json:"cash"`
Equity string `json:"equity"`
BuyingPower string `json:"buying_power"`
PortfolioValue string `json:"portfolio_value"`
Positions []types.Position `json:"positions"`
}
// GetOmnibusSnapshot returns the current state of the omnibus master
// for end-of-day reconciliation. This is the aggregate across all
// sub-accounts and should match the sum of per-sub positions + cash.
func (p *Provider) GetOmnibusSnapshot(ctx context.Context) (*OmnibusSnapshot, error) {
tData, _, err := p.do(ctx, http.MethodGet, "/v1/trading/accounts/"+p.cfg.OmnibusAccountID+"/account", nil)
if err != nil {
return nil, err
}
var ta struct {
Cash string `json:"cash"`
Equity string `json:"equity"`
BuyingPower string `json:"buying_power"`
PortfolioValue string `json:"portfolio_value"`
}
if err := json.Unmarshal(tData, &ta); err != nil {
return nil, err
}
pData, _, err := p.do(ctx, http.MethodGet, "/v1/trading/accounts/"+p.cfg.OmnibusAccountID+"/positions", nil)
if err != nil {
return nil, err
}
var rawPositions []struct {
Symbol string `json:"symbol"`
Qty string `json:"qty"`
AvgEntryPrice string `json:"avg_entry_price"`
MarketValue string `json:"market_value"`
CurrentPrice string `json:"current_price"`
UnrealizedPL string `json:"unrealized_pl"`
Side string `json:"side"`
AssetClass string `json:"asset_class"`
}
if err := json.Unmarshal(pData, &rawPositions); err != nil {
return nil, err
}
positions := make([]types.Position, len(rawPositions))
for i, rp := range rawPositions {
positions[i] = types.Position{
Symbol: rp.Symbol,
Qty: rp.Qty,
AvgEntryPrice: rp.AvgEntryPrice,
MarketValue: rp.MarketValue,
CurrentPrice: rp.CurrentPrice,
UnrealizedPL: rp.UnrealizedPL,
Side: rp.Side,
AssetClass: rp.AssetClass,
}
}
return &OmnibusSnapshot{
AccountID: p.cfg.OmnibusAccountID,
Cash: ta.Cash,
Equity: ta.Equity,
BuyingPower: ta.BuyingPower,
PortfolioValue: ta.PortfolioValue,
Positions: positions,
}, nil
}
@@ -0,0 +1,651 @@
package alpaca_omnisub
import (
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/luxfi/broker/pkg/types"
)
const testOmnibusID = "omni-master-001"
func testServer(t *testing.T, handlers map[string]http.HandlerFunc) (*httptest.Server, *Provider) {
t.Helper()
mux := http.NewServeMux()
for pattern, handler := range handlers {
mux.HandleFunc(pattern, handler)
}
srv := httptest.NewServer(mux)
t.Cleanup(srv.Close)
p := New(Config{
BaseURL: srv.URL,
APIKey: "test-key",
APISecret: "test-secret",
OmnibusAccountID: testOmnibusID,
})
return srv, p
}
func testDataServer(t *testing.T, p *Provider, handlers map[string]http.HandlerFunc) *httptest.Server {
t.Helper()
mux := http.NewServeMux()
for pattern, handler := range handlers {
mux.HandleFunc(pattern, handler)
}
srv := httptest.NewServer(mux)
t.Cleanup(srv.Close)
p.dataURL = srv.URL
return srv
}
func TestName(t *testing.T) {
p := New(Config{})
if p.Name() != "alpaca_omnisub" {
t.Fatalf("Name() = %q, want 'alpaca_omnisub'", p.Name())
}
}
func TestDefaultsToSandbox(t *testing.T) {
p := New(Config{})
if p.cfg.BaseURL != SandboxURL {
t.Fatalf("BaseURL = %q, want %q", p.cfg.BaseURL, SandboxURL)
}
if p.dataURL != DataSandboxURL {
t.Fatalf("dataURL = %q, want %q", p.dataURL, DataSandboxURL)
}
}
func TestProductionDataURL(t *testing.T) {
p := New(Config{BaseURL: ProductionURL})
if p.dataURL != DataURL {
t.Fatalf("dataURL = %q, want %q", p.dataURL, DataURL)
}
}
func TestCreateSubAccount(t *testing.T) {
var gotBody map[string]interface{}
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/accounts": func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
t.Errorf("method = %s, want POST", r.Method)
}
user, pass, ok := r.BasicAuth()
if !ok || user != "test-key" || pass != "test-secret" {
t.Error("missing or wrong basic auth")
}
body, _ := io.ReadAll(r.Body)
json.Unmarshal(body, &gotBody)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"id": "sub-uuid-1",
"account_number": "SUB001",
"status": "ACTIVE",
"currency": "USD",
"account_type": "omnibus_sub_account",
"created_at": "2026-04-01T10:00:00Z",
"identity": map[string]interface{}{
"given_name": "Alice",
"family_name": "Smith",
},
"contact": map[string]interface{}{
"email_address": "[email protected]",
},
})
},
})
ctx := context.Background()
acct, err := p.CreateAccount(ctx, &types.CreateAccountRequest{
Identity: &types.Identity{
GivenName: "Alice",
FamilyName: "Smith",
TaxID: "123-45-6789",
},
Contact: &types.Contact{
Email: "[email protected]",
Phone: "555-0001",
},
IPAddress: "203.0.113.10",
})
if err != nil {
t.Fatalf("CreateAccount error: %v", err)
}
if acct.ProviderID != "sub-uuid-1" {
t.Errorf("ProviderID = %q, want 'sub-uuid-1'", acct.ProviderID)
}
if acct.Provider != "alpaca_omnisub" {
t.Errorf("Provider = %q, want 'alpaca_omnisub'", acct.Provider)
}
if acct.AccountType != "omnibus_sub_account" {
t.Errorf("AccountType = %q, want 'omnibus_sub_account'", acct.AccountType)
}
// Verify body includes omnisub fields
if gotBody["account_type"] != "omnibus_sub_account" {
t.Errorf("body account_type = %v, want 'omnibus_sub_account'", gotBody["account_type"])
}
if gotBody["omnibus_master_id"] != testOmnibusID {
t.Errorf("body omnibus_master_id = %v, want %q", gotBody["omnibus_master_id"], testOmnibusID)
}
}
func TestCreateSubAccountMissingIP(t *testing.T) {
_, p := testServer(t, nil)
_, err := p.CreateAccount(context.Background(), &types.CreateAccountRequest{
Identity: &types.Identity{GivenName: "Bob"},
Contact: &types.Contact{Email: "[email protected]"},
})
if err == nil {
t.Fatal("expected error for missing IP")
}
if !strings.Contains(err.Error(), "IP address") {
t.Errorf("error = %q, want to contain 'IP address'", err.Error())
}
}
func TestGetAccount(t *testing.T) {
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/accounts/sub-1": func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
t.Errorf("method = %s, want GET", r.Method)
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"id": "sub-1",
"account_number": "S100",
"status": "ACTIVE",
"currency": "USD",
"account_type": "omnibus_sub_account",
"created_at": "2026-04-01T10:00:00Z",
})
},
})
acct, err := p.GetAccount(context.Background(), "sub-1")
if err != nil {
t.Fatalf("GetAccount error: %v", err)
}
if acct.ProviderID != "sub-1" {
t.Errorf("ProviderID = %q, want 'sub-1'", acct.ProviderID)
}
if acct.AccountNumber != "S100" {
t.Errorf("AccountNumber = %q, want 'S100'", acct.AccountNumber)
}
}
func TestListAccounts(t *testing.T) {
var gotQuery string
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/accounts": func(w http.ResponseWriter, r *http.Request) {
gotQuery = r.URL.RawQuery
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode([]map[string]interface{}{
{"id": "s1", "status": "ACTIVE", "account_type": "omnibus_sub_account", "created_at": "2026-04-01T10:00:00Z"},
{"id": "s2", "status": "ACTIVE", "account_type": "omnibus_sub_account", "created_at": "2026-04-02T10:00:00Z"},
})
},
})
accts, err := p.ListAccounts(context.Background())
if err != nil {
t.Fatalf("ListAccounts error: %v", err)
}
if len(accts) != 2 {
t.Fatalf("expected 2 accounts, got %d", len(accts))
}
// Verify query filters by omnibus master
if !strings.Contains(gotQuery, testOmnibusID) {
t.Errorf("query = %q, want to contain omnibus master ID %q", gotQuery, testOmnibusID)
}
}
func TestPlaceOrder(t *testing.T) {
var gotBody map[string]interface{}
var gotPath string
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/trading/accounts/sub-1/orders": func(w http.ResponseWriter, r *http.Request) {
gotPath = r.URL.Path
body, _ := io.ReadAll(r.Body)
json.Unmarshal(body, &gotBody)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"id": "order-uuid-1",
"symbol": "AAPL",
"qty": "10",
"side": "buy",
"type": "limit",
"time_in_force": "day",
"limit_price": "150.00",
"status": "accepted",
"created_at": "2026-04-01T10:00:00Z",
})
},
})
order, err := p.CreateOrder(context.Background(), "sub-1", &types.CreateOrderRequest{
Symbol: "AAPL",
Qty: "10",
Side: "buy",
Type: "limit",
TimeInForce: "day",
LimitPrice: "150.00",
})
if err != nil {
t.Fatalf("CreateOrder error: %v", err)
}
if gotPath != "/v1/trading/accounts/sub-1/orders" {
t.Errorf("path = %s, want /v1/trading/accounts/sub-1/orders", gotPath)
}
if order.ProviderID != "order-uuid-1" {
t.Errorf("ProviderID = %q, want 'order-uuid-1'", order.ProviderID)
}
if order.Provider != "alpaca_omnisub" {
t.Errorf("Provider = %q, want 'alpaca_omnisub'", order.Provider)
}
if order.Status != "accepted" {
t.Errorf("Status = %q, want 'accepted'", order.Status)
}
if gotBody["symbol"] != "AAPL" {
t.Errorf("body symbol = %v, want 'AAPL'", gotBody["symbol"])
}
}
func TestGetOrder(t *testing.T) {
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/trading/accounts/sub-1/orders/o1": func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
t.Errorf("method = %s, want GET", r.Method)
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"id": "o1",
"symbol": "AAPL",
"status": "filled",
"created_at": "2026-04-01T10:00:00Z",
})
},
})
order, err := p.GetOrder(context.Background(), "sub-1", "o1")
if err != nil {
t.Fatalf("GetOrder error: %v", err)
}
if order.ProviderID != "o1" {
t.Errorf("ProviderID = %q, want 'o1'", order.ProviderID)
}
}
func TestCancelOrder(t *testing.T) {
var gotMethod string
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/trading/accounts/sub-1/orders/o1": func(w http.ResponseWriter, r *http.Request) {
gotMethod = r.Method
w.WriteHeader(http.StatusNoContent)
},
})
err := p.CancelOrder(context.Background(), "sub-1", "o1")
if err != nil {
t.Fatalf("CancelOrder error: %v", err)
}
if gotMethod != http.MethodDelete {
t.Errorf("method = %s, want DELETE", gotMethod)
}
}
func TestListOrders(t *testing.T) {
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/trading/accounts/sub-1/orders": func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode([]map[string]interface{}{
{"id": "o1", "symbol": "AAPL", "status": "filled", "created_at": "2026-04-01T10:00:00Z"},
{"id": "o2", "symbol": "GOOG", "status": "accepted", "created_at": "2026-04-01T11:00:00Z"},
})
},
})
orders, err := p.ListOrders(context.Background(), "sub-1")
if err != nil {
t.Fatalf("ListOrders error: %v", err)
}
if len(orders) != 2 {
t.Fatalf("expected 2 orders, got %d", len(orders))
}
}
func TestGetPortfolio(t *testing.T) {
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/trading/accounts/sub-1/account": func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"cash": "5000.00",
"equity": "25000.00",
"buying_power": "50000.00",
"portfolio_value": "25000.00",
})
},
"/v1/trading/accounts/sub-1/positions": func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode([]map[string]interface{}{
{
"symbol": "AAPL",
"qty": "20",
"avg_entry_price": "150.00",
"market_value": "3100.00",
"current_price": "155.00",
"unrealized_pl": "100.00",
"side": "long",
"asset_class": "us_equity",
},
})
},
})
port, err := p.GetPortfolio(context.Background(), "sub-1")
if err != nil {
t.Fatalf("GetPortfolio error: %v", err)
}
if port.Cash != "5000.00" {
t.Errorf("Cash = %q, want '5000.00'", port.Cash)
}
if len(port.Positions) != 1 {
t.Fatalf("expected 1 position, got %d", len(port.Positions))
}
if port.Positions[0].Symbol != "AAPL" {
t.Errorf("position symbol = %q, want 'AAPL'", port.Positions[0].Symbol)
}
}
func TestTransferGoesToOmnibus(t *testing.T) {
var gotPath string
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/accounts/" + testOmnibusID + "/transfers": func(w http.ResponseWriter, r *http.Request) {
gotPath = r.URL.Path
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"id": "xfer-1",
"account_id": testOmnibusID,
"type": "ach",
"direction": "INCOMING",
"amount": "1000.00",
"currency": "USD",
"status": "QUEUED",
"created_at": "2026-04-01T10:00:00Z",
"updated_at": "2026-04-01T10:00:00Z",
})
},
})
xfer, err := p.CreateTransfer(context.Background(), "sub-1", &types.CreateTransferRequest{
Type: "ach",
Direction: "INCOMING",
Amount: "1000.00",
})
if err != nil {
t.Fatalf("CreateTransfer error: %v", err)
}
// Verify the transfer went to the omnibus, not the sub
if gotPath != "/v1/accounts/"+testOmnibusID+"/transfers" {
t.Errorf("path = %s, want /v1/accounts/%s/transfers", gotPath, testOmnibusID)
}
if xfer.ProviderID != "xfer-1" {
t.Errorf("ProviderID = %q, want 'xfer-1'", xfer.ProviderID)
}
if xfer.Provider != "alpaca_omnisub" {
t.Errorf("Provider = %q, want 'alpaca_omnisub'", xfer.Provider)
}
}
func TestListTransfersOnOmnibus(t *testing.T) {
var gotPath string
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/accounts/" + testOmnibusID + "/transfers": func(w http.ResponseWriter, r *http.Request) {
gotPath = r.URL.Path
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode([]map[string]interface{}{
{"id": "x1", "amount": "500.00", "status": "COMPLETE", "created_at": "2026-04-01T10:00:00Z", "updated_at": "2026-04-01T10:00:00Z"},
})
},
})
xfers, err := p.ListTransfers(context.Background(), "sub-1")
if err != nil {
t.Fatalf("ListTransfers error: %v", err)
}
if len(xfers) != 1 {
t.Fatalf("expected 1 transfer, got %d", len(xfers))
}
if gotPath != "/v1/accounts/"+testOmnibusID+"/transfers" {
t.Errorf("path = %s, want omnibus transfers path", gotPath)
}
}
func TestCreateJournal(t *testing.T) {
var gotBody map[string]interface{}
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/journals": func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
t.Errorf("method = %s, want POST", r.Method)
}
body, _ := io.ReadAll(r.Body)
json.Unmarshal(body, &gotBody)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"id": "j-1",
"entry_type": "JNLC",
"from_account": testOmnibusID,
"to_account": "sub-1",
"net_amount": "500.00",
"status": "executed",
"created_at": "2026-04-01T10:00:00Z",
})
},
})
j, err := p.CreateJournal(context.Background(), &types.CreateJournalRequest{
EntryType: "JNLC",
FromAccount: testOmnibusID,
ToAccount: "sub-1",
Amount: "500.00",
Description: "fund sub-account",
})
if err != nil {
t.Fatalf("CreateJournal error: %v", err)
}
if j.ID != "j-1" {
t.Errorf("ID = %q, want 'j-1'", j.ID)
}
if j.Amount != "500.00" {
t.Errorf("Amount = %q, want '500.00'", j.Amount)
}
if gotBody["from_account"] != testOmnibusID {
t.Errorf("body from_account = %v, want %q", gotBody["from_account"], testOmnibusID)
}
if gotBody["description"] != "fund sub-account" {
t.Errorf("body description = %v, want 'fund sub-account'", gotBody["description"])
}
}
func TestGetOmnibusSnapshot(t *testing.T) {
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/trading/accounts/" + testOmnibusID + "/account": func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"cash": "100000.00",
"equity": "500000.00",
"buying_power": "1000000.00",
"portfolio_value": "500000.00",
})
},
"/v1/trading/accounts/" + testOmnibusID + "/positions": func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode([]map[string]interface{}{
{
"symbol": "AAPL",
"qty": "1000",
"avg_entry_price": "150.00",
"market_value": "155000.00",
"current_price": "155.00",
"unrealized_pl": "5000.00",
"side": "long",
"asset_class": "us_equity",
},
{
"symbol": "GOOG",
"qty": "200",
"avg_entry_price": "2800.00",
"market_value": "580000.00",
"current_price": "2900.00",
"unrealized_pl": "20000.00",
"side": "long",
"asset_class": "us_equity",
},
})
},
})
snap, err := p.GetOmnibusSnapshot(context.Background())
if err != nil {
t.Fatalf("GetOmnibusSnapshot error: %v", err)
}
if snap.AccountID != testOmnibusID {
t.Errorf("AccountID = %q, want %q", snap.AccountID, testOmnibusID)
}
if snap.Cash != "100000.00" {
t.Errorf("Cash = %q, want '100000.00'", snap.Cash)
}
if snap.Equity != "500000.00" {
t.Errorf("Equity = %q, want '500000.00'", snap.Equity)
}
if len(snap.Positions) != 2 {
t.Fatalf("expected 2 positions, got %d", len(snap.Positions))
}
if snap.Positions[0].Symbol != "AAPL" {
t.Errorf("positions[0].Symbol = %q, want 'AAPL'", snap.Positions[0].Symbol)
}
}
func TestBankRelationshipOnOmnibus(t *testing.T) {
var gotPath string
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/accounts/" + testOmnibusID + "/ach_relationships": func(w http.ResponseWriter, r *http.Request) {
gotPath = r.URL.Path
w.Header().Set("Content-Type", "application/json")
if r.Method == http.MethodPost {
json.NewEncoder(w).Encode(map[string]interface{}{
"id": "ach-1",
"account_id": testOmnibusID,
"account_owner_name": "Omnibus LLC",
"bank_account_type": "CHECKING",
"status": "APPROVED",
})
} else {
json.NewEncoder(w).Encode([]map[string]interface{}{
{
"id": "ach-1",
"account_id": testOmnibusID,
"account_owner_name": "Omnibus LLC",
"bank_account_type": "CHECKING",
"status": "APPROVED",
},
})
}
},
})
// Create
br, err := p.CreateBankRelationship(context.Background(), "sub-1", "Omnibus LLC", "CHECKING", "123456789", "021000021")
if err != nil {
t.Fatalf("CreateBankRelationship error: %v", err)
}
if br.ProviderID != "ach-1" {
t.Errorf("ProviderID = %q, want 'ach-1'", br.ProviderID)
}
if gotPath != "/v1/accounts/"+testOmnibusID+"/ach_relationships" {
t.Errorf("path = %s, want omnibus ach path", gotPath)
}
// List
rels, err := p.ListBankRelationships(context.Background(), "sub-1")
if err != nil {
t.Fatalf("ListBankRelationships error: %v", err)
}
if len(rels) != 1 {
t.Fatalf("expected 1 relationship, got %d", len(rels))
}
}
func TestGetSnapshotStock(t *testing.T) {
_, p := testServer(t, nil)
testDataServer(t, p, map[string]http.HandlerFunc{
"/v2/stocks/AAPL/snapshot": func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"latestTrade": map[string]interface{}{
"t": "2026-04-01T10:00:00Z",
"p": 155.0,
"s": 100.0,
"x": "V",
},
"latestQuote": map[string]interface{}{
"t": "2026-04-01T10:00:00Z",
"bp": 154.9,
"bs": 200.0,
"ap": 155.1,
"as": 150.0,
},
})
},
})
snap, err := p.GetSnapshot(context.Background(), "AAPL")
if err != nil {
t.Fatalf("GetSnapshot error: %v", err)
}
if snap.Symbol != "AAPL" {
t.Errorf("Symbol = %q, want 'AAPL'", snap.Symbol)
}
if snap.LatestTrade == nil || snap.LatestTrade.Price != 155.0 {
t.Errorf("LatestTrade.Price = %v, want 155.0", snap.LatestTrade)
}
}
func TestErrorHandling(t *testing.T) {
_, p := testServer(t, map[string]http.HandlerFunc{
"/v1/accounts/bad-id": func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusNotFound)
json.NewEncoder(w).Encode(map[string]interface{}{
"code": 40410000,
"message": "account not found",
})
},
})
_, err := p.GetAccount(context.Background(), "bad-id")
if err == nil {
t.Fatal("expected error for 404")
}
if !strings.Contains(err.Error(), "account not found") {
t.Errorf("error = %q, want to contain 'account not found'", err.Error())
}
if !strings.Contains(err.Error(), "alpaca_omnisub") {
t.Errorf("error = %q, want to contain provider name", err.Error())
}
}
@@ -0,0 +1,146 @@
package alpaca_omnisub
import (
"context"
"encoding/json"
"net/http"
"time"
"github.com/luxfi/broker/pkg/types"
)
// CreateTransfer initiates a fund transfer on the omnibus master account.
// In the OmniSub model, ACH/wire flows through the omnibus — not the
// sub-account directly. The caller passes subAccountID for audit, but the
// actual transfer is booked on the omnibus master. Internal ledger
// credit/debit to the sub-account is handled via journals.
func (p *Provider) CreateTransfer(ctx context.Context, subAccountID string, req *types.CreateTransferRequest) (*types.Transfer, error) {
body := map[string]interface{}{
"transfer_type": req.Type,
"amount": req.Amount,
"direction": req.Direction,
}
if req.RelationshipID != "" {
body["relationship_id"] = req.RelationshipID
}
// Transfers go through the omnibus master.
data, _, err := p.do(ctx, http.MethodPost, "/v1/accounts/"+p.cfg.OmnibusAccountID+"/transfers", body)
if err != nil {
return nil, err
}
return p.parseTransfer(data)
}
// ListTransfers lists transfers on the omnibus master.
// Per the OmniSub model, all external fund movements are on the omnibus.
func (p *Provider) ListTransfers(ctx context.Context, _ string) ([]*types.Transfer, error) {
data, _, err := p.do(ctx, http.MethodGet, "/v1/accounts/"+p.cfg.OmnibusAccountID+"/transfers", nil)
if err != nil {
return nil, err
}
var raw []json.RawMessage
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
transfers := make([]*types.Transfer, 0, len(raw))
for _, r := range raw {
t, err := p.parseTransfer(r)
if err != nil {
continue
}
transfers = append(transfers, t)
}
return transfers, nil
}
// CreateBankRelationship creates an ACH relationship on the omnibus master.
func (p *Provider) CreateBankRelationship(ctx context.Context, _ string, ownerName, accountType, accountNumber, routingNumber string) (*types.BankRelationship, error) {
body := map[string]interface{}{
"account_owner_name": ownerName,
"bank_account_type": accountType,
"bank_account_number": accountNumber,
"bank_routing_number": routingNumber,
}
data, _, err := p.do(ctx, http.MethodPost, "/v1/accounts/"+p.cfg.OmnibusAccountID+"/ach_relationships", body)
if err != nil {
return nil, err
}
return p.parseBankRelationship(data)
}
// ListBankRelationships lists ACH relationships on the omnibus master.
func (p *Provider) ListBankRelationships(ctx context.Context, _ string) ([]*types.BankRelationship, error) {
data, _, err := p.do(ctx, http.MethodGet, "/v1/accounts/"+p.cfg.OmnibusAccountID+"/ach_relationships", nil)
if err != nil {
return nil, err
}
var raw []json.RawMessage
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
rels := make([]*types.BankRelationship, 0, len(raw))
for _, r := range raw {
br, err := p.parseBankRelationship(r)
if err != nil {
continue
}
rels = append(rels, br)
}
return rels, nil
}
func (p *Provider) parseTransfer(data []byte) (*types.Transfer, error) {
var raw struct {
ID string `json:"id"`
AccountID string `json:"account_id"`
Type string `json:"type"`
Status string `json:"status"`
Amount string `json:"amount"`
Direction string `json:"direction"`
Currency string `json:"currency"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
t := &types.Transfer{
Provider: "alpaca_omnisub",
ProviderID: raw.ID,
AccountID: raw.AccountID,
Type: raw.Type,
Direction: raw.Direction,
Amount: raw.Amount,
Currency: raw.Currency,
Status: raw.Status,
}
if ts, err := time.Parse(time.RFC3339Nano, raw.CreatedAt); err == nil {
t.CreatedAt = ts
}
if ts, err := time.Parse(time.RFC3339Nano, raw.UpdatedAt); err == nil {
t.UpdatedAt = ts
}
return t, nil
}
func (p *Provider) parseBankRelationship(data []byte) (*types.BankRelationship, error) {
var raw struct {
ID string `json:"id"`
AccountID string `json:"account_id"`
AccountOwnerName string `json:"account_owner_name"`
BankAccountType string `json:"bank_account_type"`
Status string `json:"status"`
}
if err := json.Unmarshal(data, &raw); err != nil {
return nil, err
}
return &types.BankRelationship{
Provider: "alpaca_omnisub",
ProviderID: raw.ID,
AccountID: raw.AccountID,
AccountOwnerName: raw.AccountOwnerName,
BankAccountType: raw.BankAccountType,
Status: raw.Status,
}, nil
}
+87 -1
View File
@@ -9,6 +9,7 @@ import (
"github.com/luxfi/broker/pkg/provider"
"github.com/luxfi/broker/pkg/provider/apex"
"github.com/luxfi/broker/pkg/provider/alpaca"
"github.com/luxfi/broker/pkg/provider/alpaca_omnisub"
"github.com/luxfi/broker/pkg/provider/binance"
"github.com/luxfi/broker/pkg/provider/bitgo"
"github.com/luxfi/broker/pkg/provider/circle"
@@ -25,6 +26,14 @@ import (
"github.com/luxfi/broker/pkg/provider/polygon"
"github.com/luxfi/broker/pkg/provider/sfox"
"github.com/luxfi/broker/pkg/provider/tradier"
// International adapters
"github.com/luxfi/broker/pkg/provider/btgpactual"
"github.com/luxfi/broker/pkg/provider/ig"
"github.com/luxfi/broker/pkg/provider/questrade"
"github.com/luxfi/broker/pkg/provider/saxo"
"github.com/luxfi/broker/pkg/provider/stake"
"github.com/luxfi/broker/pkg/provider/zerodha"
)
// RegisterFromEnv reads provider environment variables and registers
@@ -43,7 +52,6 @@ func RegisterFromEnv(registry *provider.Registry) int {
DEXAddr: envOr("LX_DEX_ADDR", lx.DefaultDEXAddr),
MPCAddr: envOr("LX_MPC_ADDR", lx.DefaultMPCAddr),
USDLAddress: os.Getenv("LX_USDL_ADDR"),
RPCURL: os.Getenv("LX_RPC_URL"),
}))
slog.Info("provider registered", "name", "lx", "transport", "zap")
n++
@@ -59,6 +67,17 @@ func RegisterFromEnv(registry *provider.Registry) int {
n++
}
if key := os.Getenv("ALPACA_OMNISUB_API_KEY"); key != "" {
registry.Register(alpaca_omnisub.New(alpaca_omnisub.Config{
BaseURL: envOr("ALPACA_OMNISUB_BASE_URL", alpaca_omnisub.SandboxURL),
APIKey: key,
APISecret: os.Getenv("ALPACA_OMNISUB_API_SECRET"),
OmnibusAccountID: os.Getenv("ALPACA_OMNISUB_OMNIBUS_ACCOUNT_ID"),
}))
slog.Info("provider registered", "name", "alpaca_omnisub")
n++
}
if token := os.Getenv("IBKR_ACCESS_TOKEN"); token != "" {
registry.Register(ibkr.New(ibkr.Config{
GatewayURL: envOr("IBKR_GATEWAY_URL", ibkr.DefaultGatewayURL),
@@ -230,6 +249,73 @@ func RegisterFromEnv(registry *provider.Registry) int {
n++
}
// --- International adapters ---
if id := os.Getenv("BTG_CLIENT_ID"); id != "" {
registry.Register(btgpactual.New(btgpactual.Config{
BaseURL: envOr("BTG_BASE_URL", btgpactual.SandboxURL),
ClientID: id,
ClientSecret: os.Getenv("BTG_CLIENT_SECRET"),
AccountID: os.Getenv("BTG_ACCOUNT_ID"),
}))
slog.Info("provider registered", "name", "btgpactual")
n++
}
if key := os.Getenv("IG_API_KEY"); key != "" {
registry.Register(ig.New(ig.Config{
BaseURL: envOr("IG_BASE_URL", ig.DemoURL),
APIKey: key,
Username: os.Getenv("IG_USERNAME"),
Password: os.Getenv("IG_PASSWORD"),
AccountID: os.Getenv("IG_ACCOUNT_ID"),
}))
slog.Info("provider registered", "name", "ig")
n++
}
if key := os.Getenv("ZERODHA_API_KEY"); key != "" {
registry.Register(zerodha.New(zerodha.Config{
BaseURL: envOr("ZERODHA_BASE_URL", zerodha.ProdURL),
APIKey: key,
APISecret: os.Getenv("ZERODHA_API_SECRET"),
}))
slog.Info("provider registered", "name", "zerodha")
n++
}
if token := os.Getenv("SAXO_ACCESS_TOKEN"); token != "" {
registry.Register(saxo.New(saxo.Config{
BaseURL: envOr("SAXO_BASE_URL", saxo.SimURL),
AccessToken: token,
ClientID: os.Getenv("SAXO_CLIENT_ID"),
ClientSecret: os.Getenv("SAXO_CLIENT_SECRET"),
AccountKey: os.Getenv("SAXO_ACCOUNT_KEY"),
}))
slog.Info("provider registered", "name", "saxo")
n++
}
if token := os.Getenv("QUESTRADE_REFRESH_TOKEN"); token != "" {
registry.Register(questrade.New(questrade.Config{
BaseURL: envOr("QUESTRADE_BASE_URL", questrade.ProdURL),
RefreshToken: token,
AccountID: os.Getenv("QUESTRADE_ACCOUNT_ID"),
}))
slog.Info("provider registered", "name", "questrade")
n++
}
if key := os.Getenv("STAKE_API_KEY"); key != "" {
registry.Register(stake.New(stake.Config{
BaseURL: envOr("STAKE_BASE_URL", stake.ProdURL),
APIKey: key,
APISecret: os.Getenv("STAKE_API_SECRET"),
}))
slog.Info("provider registered", "name", "stake")
n++
}
return n
}