fisk/core/lib/schemas/oanda_s.py

from decimal import Decimal as D
from typing import Optional

from pydantic import BaseModel


class PositionLong(BaseModel):
    units: str
    averagePrice: Optional[str] = None
    pl: str
    resettablePL: str
    financing: str
    dividendAdjustment: str
    guaranteedExecutionFees: str
    tradeIDs: Optional[list[str]] = []
    unrealizedPL: str


class PositionShort(BaseModel):
    units: str
    averagePrice: Optional[str] = None
    pl: str
    resettablePL: str
    financing: str
    dividendAdjustment: str
    guaranteedExecutionFees: str
    tradeIDs: Optional[list[str]] = []
    unrealizedPL: str


class Position(BaseModel):
    instrument: str
    long: PositionLong
    short: PositionShort
    pl: str
    resettablePL: str
    financing: str
    commission: str
    dividendAdjustment: str
    guaranteedExecutionFees: str
    unrealizedPL: str
    marginUsed: str


class OpenPositions(BaseModel):
    positions: list[Position]
    lastTransactionID: str


def parse_time(x):
    """
    Parse the time from the Oanda API.
    """
    if "openTime" in x:
        ts_split = x["openTime"].split(".")
    else:
        ts_split = x["trade"]["openTime"].split(".")
    microseconds = ts_split[1].replace("Z", "")
    microseconds_6 = microseconds[:6]
    new_ts = ts_split[0] + "." + microseconds_6 + "Z"
    return new_ts


def prevent_hedging(x):
    """
    Our implementation breaks if a position has both.
    We implemented it this way in order to more easily support other exchanges.
    The actual direction is put into the root object with Grom.
    """
    if float(x["long"]["units"]) > 0 and float(x["short"]["units"]) < 0:
        raise ValueError("Hedging not allowed")


def parse_prices(x):
    prevent_hedging(x)
    if float(x["long"]["units"]) > 0:
        return x["long"]["averagePrice"]
    elif float(x["short"]["units"]) < 0:
        return x["short"]["averagePrice"]
    else:
        return 0


def parse_units(x):
    prevent_hedging(x)
    if float(x["long"]["units"]) > 0:
        return x["long"]["units"]
    elif float(x["short"]["units"]) < 0:
        return x["short"]["units"]
    else:
        return 0


def parse_value(x):
    prevent_hedging(x)
    if float(x["long"]["units"]) > 0:
        return D(x["long"]["units"]) * D(x["long"]["averagePrice"])
    elif float(x["short"]["units"]) < 0:
        return D(x["short"]["units"]) * D(x["short"]["averagePrice"])
    else:
        return 0


def parse_current_units_side(x):
    if float(x["currentUnits"]) > 0:
        return "long"
    elif float(x["currentUnits"]) < 0:
        return "short"


def parse_side(x):
    prevent_hedging(x)
    if float(x["long"]["units"]) > 0:
        return "long"
    elif float(x["short"]["units"]) < 0:
        return "short"
    else:
        return "unknown"


def parse_trade_ids(x, sum=0):
    prevent_hedging(x)
    if float(x["long"]["units"]) > 0:
        return [str(int(y) + sum) for y in x["long"]["tradeIDs"]]
    elif float(x["short"]["units"]) < 0:
        return [str(int(y) + sum) for y in x["short"]["tradeIDs"]]
    else:
        return "unknown"


OpenPositionsSchema = {
    "itemlist": (
        "positions",
        [
            {
                "symbol": "instrument",
                "unrealized_pl": "unrealizedPL",
                "trade_ids": parse_trade_ids,  # actual value is lower by 1
                "price": parse_prices,
                "units": parse_units,
                "side": parse_side,
                "value": parse_value,
            }
        ],
    )
}


class AccountDetailsNested(BaseModel):
    guaranteedStopLossOrderMode: str
    hedgingEnabled: bool
    id: str
    createdTime: str
    currency: str
    createdByUserID: int
    alias: str
    marginRate: str
    lastTransactionID: str
    balance: str
    openTradeCount: int
    openPositionCount: int
    pendingOrderCount: int
    pl: str
    resettablePL: str
    resettablePLTime: str
    financing: str
    commission: str
    dividendAdjustment: str
    guaranteedExecutionFees: str
    orders: list  # Order
    positions: list  # Position
    trades: list  # Trade
    unrealizedPL: str
    NAV: str
    marginUsed: str
    marginAvailable: str
    positionValue: str
    marginCloseoutUnrealizedPL: str
    marginCloseoutNAV: str
    marginCloseoutMarginUsed: str
    marginCloseoutPositionValue: str
    marginCloseoutPercent: str
    withdrawalLimit: str
    marginCallMarginUsed: str
    marginCallPercent: str


class AccountDetails(BaseModel):
    account: AccountDetailsNested
    lastTransactionID: str


AccountDetailsSchema = {
    "guaranteedSLOM": "account.guaranteedStopLossOrderMode",
    "hedgingEnabled": "account.hedgingEnabled",
    "id": "account.id",
    "created_at": "account.createdTime",
    "currency": "account.currency",
    "createdByUserID": "account.createdByUserID",
    "alias": "account.alias",
    "marginRate": "account.marginRate",
    "lastTransactionID": "account.lastTransactionID",
    "balance": "account.balance",
    "openTradeCount": "account.openTradeCount",
    "openPositionCount": "account.openPositionCount",
    "pendingOrderCount": "account.pendingOrderCount",
    "pl": "account.pl",
    "resettablePL": "account.resettablePL",
    "resettablePLTime": "account.resettablePLTime",
    "financing": "account.financing",
    "commission": "account.commission",
    "dividendAdjustment": "account.dividendAdjustment",
    "guaranteedExecutionFees": "account.guaranteedExecutionFees",
    # "orders": "account.orders",
    # "positions": "account.positions",
    # "trades": "account.trades",
    "unrealizedPL": "account.unrealizedPL",
    "NAV": "account.NAV",
    "marginUsed": "account.marginUsed",
    "marginAvailable": "account.marginAvailable",
    "positionValue": "account.positionValue",
    "marginCloseoutUnrealizedPL": "account.marginCloseoutUnrealizedPL",
    "marginCloseoutNAV": "account.marginCloseoutNAV",
    "marginCloseoutMarginUsed": "account.marginCloseoutMarginUsed",
    "marginCloseoutPositionValue": "account.marginCloseoutPositionValue",
    "marginCloseoutPercent": "account.marginCloseoutPercent",
    "withdrawalLimit": "account.withdrawalLimit",
    "marginCallMarginUsed": "account.marginCallMarginUsed",
    "marginCallPercent": "account.marginCallPercent",
}


class AccountSummaryNested(BaseModel):
    marginCloseoutNAV: str
    marginUsed: str
    currency: str
    resettablePL: str
    NAV: str
    marginCloseoutMarginUsed: str
    marginCloseoutPositionValue: str
    openTradeCount: int
    id: str
    hedgingEnabled: bool
    marginCloseoutPercent: str
    marginCallMarginUsed: str
    openPositionCount: int
    positionValue: str
    pl: str
    lastTransactionID: str
    marginAvailable: str
    marginRate: str
    marginCallPercent: str
    pendingOrderCount: int
    withdrawalLimit: str
    unrealizedPL: str
    alias: str
    createdByUserID: int
    marginCloseoutUnrealizedPL: str
    createdTime: str
    balance: str


class AccountSummary(BaseModel):
    account: AccountSummaryNested
    lastTransactionID: str


AccountSummarySchema = {
    "marginCloseoutNAV": "account.marginCloseoutNAV",
    "marginUsed": "account.marginUsed",
    "currency": "account.currency",
    "resettablePL": "account.resettablePL",
    "NAV": "account.NAV",
    "marginCloseoutMarginUsed": "account.marginCloseoutMarginUsed",
    "marginCloseoutPositionValue": "account.marginCloseoutPositionValue",
    "openTradeCount": "account.openTradeCount",
    "id": "account.id",
    "hedgingEnabled": "account.hedgingEnabled",
    "marginCloseoutPercent": "account.marginCloseoutPercent",
    "marginCallMarginUsed": "account.marginCallMarginUsed",
    "openPositionCount": "account.openPositionCount",
    "positionValue": "account.positionValue",
    "pl": "account.pl",
    "lastTransactionID": "account.lastTransactionID",
    "marginAvailable": "account.marginAvailable",
    "marginRate": "account.marginRate",
    "marginCallPercent": "account.marginCallPercent",
    "pendingOrderCount": "account.pendingOrderCount",
    "withdrawalLimit": "account.withdrawalLimit",
    "unrealizedPL": "account.unrealizedPL",
    "alias": "account.alias",
    "createdByUserID": "account.createdByUserID",
    "marginCloseoutUnrealizedPL": "account.marginCloseoutUnrealizedPL",
    "createdTime": "account.createdTime",
    "balance": "account.balance",
}


class PositionDetailsNested(BaseModel):
    instrument: str
    long: PositionLong
    short: PositionShort
    pl: str
    resettablePL: str
    financing: str
    commission: str
    dividendAdjustment: str
    guaranteedExecutionFees: str
    unrealizedPL: str
    marginUsed: Optional[str] = None


class PositionDetails(BaseModel):
    position: PositionDetailsNested
    lastTransactionID: str


PositionDetailsSchema = {
    "symbol": "position.instrument",
    "long": "position.long",
    "short": "position.short",
    "pl": "position.pl",
    "resettablePL": "position.resettablePL",
    "financing": "position.financing",
    "commission": "position.commission",
    "dividendAdjustment": "position.dividendAdjustment",
    "guaranteedExecutionFees": "position.guaranteedExecutionFees",
    "unrealizedPL": "position.unrealizedPL",
    "marginUsed": "position.marginUsed",
    "price": lambda x: parse_prices(x["position"]),
    "units": lambda x: parse_units(x["position"]),
    "side": lambda x: parse_side(x["position"]),
    "value": lambda x: parse_value(x["position"]),
    "trade_ids": lambda x: parse_trade_ids(
        x["position"], sum=0
    ),  # this value is correct
}


class InstrumentTag(BaseModel):
    type: str
    name: str


class InstrumentFinancingDaysOfWeek(BaseModel):
    dayOfWeek: str
    daysCharged: int


class InstrumentFinancing(BaseModel):
    longRate: str
    shortRate: str
    financingDaysOfWeek: list[InstrumentFinancingDaysOfWeek]


class InstrumentGuaranteedRestriction(BaseModel):
    volume: str
    priceRange: str


class Instrument(BaseModel):
    name: str
    type: str
    displayName: str
    pipLocation: int
    displayPrecision: int
    tradeUnitsPrecision: int
    minimumTradeSize: str
    maximumTrailingStopDistance: str
    minimumTrailingStopDistance: str
    maximumPositionSize: str
    maximumOrderUnits: str
    marginRate: str
    guaranteedStopLossOrderMode: str
    tags: list[InstrumentTag]
    financing: InstrumentFinancing
    guaranteedStopLossOrderLevelRestriction: Optional[
        InstrumentGuaranteedRestriction
    ] = None


class AccountInstruments(BaseModel):
    instruments: list[Instrument]


AccountInstrumentsSchema = {
    "itemlist": (
        "instruments",
        [
            {
                "name": "name",
                "type": "type",
                "displayName": "displayName",
                "pipLocation": "pipLocation",
                "displayPrecision": "displayPrecision",
                "tradeUnitsPrecision": "tradeUnitsPrecision",
                "minimumTradeSize": "minimumTradeSize",
                "maximumTrailingStopDistance": "maximumTrailingStopDistance",
                "minimumTrailingStopDistance": "minimumTrailingStopDistance",
                "maximumPositionSize": "maximumPositionSize",
                "maximumOrderUnits": "maximumOrderUnits",
                "marginRate": "marginRate",
                "guaranteedSLOM": "guaranteedStopLossOrderMode",
                "tags": "tags",
                "financing": "financing",
                "guaranteedSLOLR": "guaranteedStopLossOrderLevelRestriction",
            }
        ],
    )
}


class PriceBid(BaseModel):
    price: str
    liquidity: int


class PriceAsk(BaseModel):
    price: str
    liquidity: int


class PriceQuoteHomeConversionFactors(BaseModel):
    positiveUnits: str
    negativeUnits: str


class Price(BaseModel):
    type: str
    time: str
    bids: list[PriceBid]
    asks: list[PriceAsk]
    closeoutBid: str
    closeoutAsk: str
    status: str
    tradeable: bool
    quoteHomeConversionFactors: PriceQuoteHomeConversionFactors
    instrument: str


class PricingInfo(BaseModel):
    time: str
    prices: list[Price]


PricingInfoSchema = {
    "time": "time",
    "prices": (
        "prices",
        [
            {
                "type": "type",
                "time": "time",
                "bids": "bids",
                "asks": "asks",
                "closeoutBid": "closeoutBid",
                "closeoutAsk": "closeoutAsk",
                "status": "status",
                "tradeable": "tradeable",
                "quoteHomeConversionFactors": "quoteHomeConversionFactors",
                "symbol": "instrument",
            }
        ],
    ),
}


class Trade(BaseModel):
    tradeID: str
    clientTradeID: str
    units: str
    realizedPL: str
    financing: str
    baseFinancing: str
    price: str
    guaranteedExecutionFee: str
    quoteGuaranteedExecutionFee: str
    halfSpreadCost: str
    # takeProfitOrder: TakeProfitOrder | None
    takeProfitOrder: Optional[dict] = None
    stopLossOrder: Optional[dict] = None
    trailingStopLossOrder: Optional[dict] = None


class SideCarOrder(BaseModel):
    id: str
    createTime: str
    state: str
    price: Optional[str] = None
    timeInForce: str
    gtdTime: Optional[str] = None
    clientExtensions: Optional[dict] = None
    tradeID: str
    clientTradeID: Optional[str] = None
    type: str
    time: Optional[str] = None
    priceBound: Optional[str] = None
    positionFill: Optional[str] = None
    reason: Optional[str] = None
    orderFillTransactionID: Optional[str] = None
    tradeOpenedID: Optional[str] = None
    tradeReducedID: Optional[str] = None
    tradeClosedIDs: Optional[list[str]] = []
    cancellingTransactionID: Optional[str] = None
    replacesOrderID: Optional[str] = None
    replacedByOrderID: Optional[str] = None


class OpenTradesTrade(BaseModel):
    id: str
    instrument: str
    price: str
    openTime: str
    initialUnits: str
    initialMarginRequired: str
    state: str
    currentUnits: str
    realizedPL: str
    financing: str
    dividendAdjustment: str
    unrealizedPL: str
    marginUsed: str
    takeProfitOrder: Optional[SideCarOrder] = None
    stopLossOrder: Optional[SideCarOrder] = None
    trailingStopLossOrder: Optional[SideCarOrder] = None
    trailingStopValue: Optional[dict] = None


class OpenTrades(BaseModel):
    trades: list[OpenTradesTrade]
    lastTransactionID: str


OpenTradesSchema = {
    "itemlist": (
        "trades",
        [
            {
                "id": "id",
                "symbol": "instrument",
                "price": "price",
                "openTime": parse_time,
                "initialUnits": "initialUnits",
                "initialMarginRequired": "initialMarginRequired",
                "state": "state",
                "currentUnits": "currentUnits",
                "realizedPL": "realizedPL",
                "financing": "financing",
                "dividendAdjustment": "dividendAdjustment",
                "unrealizedPL": "unrealizedPL",
                "marginUsed": "marginUsed",
                "takeProfitOrder": "takeProfitOrder",
                "stopLossOrder": "stopLossOrder",
                "trailingStopLossOrder": "trailingStopLossOrder",
                "trailingStopValue": "trailingStopValue",
                "side": parse_current_units_side,
            }
        ],
    ),
    "lastTransactionID": "lastTransactionID",
}


class HomeConversionFactors(BaseModel):
    gainQuoteHome: str
    lossQuoteHome: str
    gainBaseHome: str
    lossBaseHome: str


class LongPositionCloseout(BaseModel):
    instrument: str
    units: str


class OrderTransaction(BaseModel):
    id: str
    accountID: str
    userID: int
    batchID: str
    requestID: str
    time: str
    type: str
    instrument: Optional[str] = None
    units: Optional[str] = None
    timeInForce: Optional[str] = None
    positionFill: Optional[str] = None
    reason: str
    longPositionCloseout: LongPositionCloseout | None
    longOrderFillTransaction: Optional[dict] = None


class OrderCreate(BaseModel):
    orderCreateTransaction: OrderTransaction


OrderCreateSchema = {
    "id": "orderCreateTransaction.id",
    "accountID": "orderCreateTransaction.accountID",
    "userID": "orderCreateTransaction.userID",
    "batchID": "orderCreateTransaction.batchID",
    "requestID": "orderCreateTransaction.requestID",
    "time": "orderCreateTransaction.time",
    "type": "orderCreateTransaction.type",
    "symbol": "orderCreateTransaction.instrument",
    "units": "orderCreateTransaction.units",
    "timeInForce": "orderCreateTransaction.timeInForce",
    "positionFill": "orderCreateTransaction.positionFill",
    "reason": "orderCreateTransaction.reason",
}


class LongOrderFillTransaction(BaseModel):
    id: str
    accountID: str
    userID: int
    batchID: str
    requestID: str
    time: str
    type: str
    orderID: str
    instrument: str
    units: str
    requestedUnits: str
    price: str
    pl: str
    quotePL: str
    financing: str
    baseFinancing: str
    commission: str
    accountBalance: str
    gainQuoteHomeConversionFactor: str
    lossQuoteHomeConversionFactor: str
    guaranteedExecutionFee: str
    quoteGuaranteedExecutionFee: str
    halfSpreadCost: str
    fullVWAP: str
    reason: str
    tradesClosed: list[Trade]
    fullPrice: Price
    homeConversionFactors: HomeConversionFactors
    longPositionCloseout: LongPositionCloseout


class PositionClose(BaseModel):
    longOrderCreateTransaction: OrderTransaction | None
    longOrderFillTransaction: OrderTransaction | None
    longOrderCancelTransaction: OrderTransaction | None
    shortOrderCreateTransaction: OrderTransaction | None
    shortOrderFillTransaction: OrderTransaction | None
    shortOrderCancelTransaction: OrderTransaction | None
    relatedTransactionIDs: list[str]
    lastTransactionID: str


PositionCloseSchema = {
    "longOrderCreateTransaction": "longOrderCreateTransaction",
    "longOrderFillTransaction": "longOrderFillTransaction",
    "longOrderCancelTransaction": "longOrderCancelTransaction",
    "shortOrderCreateTransaction": "shortOrderCreateTransaction",
    "shortOrderFillTransaction": "shortOrderFillTransaction",
    "shortOrderCancelTransaction": "shortOrderCancelTransaction",
    "relatedTransactionIDs": "relatedTransactionIDs",
    "lastTransactionID": "lastTransactionID",
}


class ClientExtensions(BaseModel):
    id: str
    tag: str


class TradeDetailsTrade(BaseModel):
    id: str
    instrument: str
    price: str
    openTime: str
    initialUnits: str
    initialMarginRequired: str
    state: str
    currentUnits: str
    realizedPL: str
    closingTransactionIDs: Optional[list[str]] = []
    financing: str
    dividendAdjustment: str
    closeTime: Optional[str] = None
    averageClosePrice: Optional[str] = None
    clientExtensions: Optional[ClientExtensions] = None


class TradeDetails(BaseModel):
    trade: TradeDetailsTrade
    lastTransactionID: str


TradeDetailsSchema = {
    "id": "trade.id",
    "symbol": "trade.instrument",
    "price": "trade.price",
    "openTime": parse_time,
    "initialUnits": "trade.initialUnits",
    "initialMarginRequired": "trade.initialMarginRequired",
    "state": "trade.state",
    "currentUnits": "trade.currentUnits",
    "realizedPL": "trade.realizedPL",
    "closingTransactionIDs": "trade.closingTransactionIDs",
    "financing": "trade.financing",
    "dividendAdjustment": "trade.dividendAdjustment",
    "closeTime": "trade.closeTime",
    "averageClosePrice": "trade.averageClosePrice",
    "clientExtensions": "trade.clientExtensions",
    "lastTransactionID": "lastTransactionID",
}


class TradeClose(BaseModel):
    orderCreateTransaction: OrderTransaction


TradeCloseSchema = {
    "id": "orderCreateTransaction.id",
    "accountID": "orderCreateTransaction.accountID",
    "userID": "orderCreateTransaction.userID",
    "batchID": "orderCreateTransaction.batchID",
    "requestID": "orderCreateTransaction.requestID",
    "time": "orderCreateTransaction.time",
    "type": "orderCreateTransaction.type",
    "symbol": "orderCreateTransaction.instrument",
    "units": "orderCreateTransaction.units",
    "timeInForce": "orderCreateTransaction.timeInForce",
    "positionFill": "orderCreateTransaction.positionFill",
    "reason": "orderCreateTransaction.reason",
    "longPositionCloseout": "orderCreateTransaction.longPositionCloseout",
    "longOrderFillTransaction": "orderCreateTransaction.longOrderFillTransaction",
}


class TradeCRCDO(BaseModel):
    takeProfitOrderCancelTransaction: Optional[OrderTransaction]
    takeProfitOrderTransaction: Optional[OrderTransaction]
    stopLossOrderCancelTransaction: Optional[OrderTransaction]
    stopLossOrderTransaction: Optional[OrderTransaction]
    relatedTransactionIDs: list[str]
    lastTransactionID: str


TradeCRCDOSchema = {
    "takeProfitOrderCancelTransaction": "takeProfitOrderCancelTransaction",
    "takeProfitOrderTransaction": "takeProfitOrderTransaction",
    "stopLossOrderCancelTransaction": "stopLossOrderCancelTransaction",
    "stopLossOrderTransaction": "stopLossOrderTransaction",
    "relatedTransactionIDs": "relatedTransactionIDs",
    "lastTransactionID": "lastTransactionID",
}