trade

BackTestExecution

BackTestExecution(
    exchange: BackTestExchange, *args, **kwargs
)

Bases: Execution

Execution for backtesting

Source code in lettrade/exchange/backtest/trade.py

def __init__(self, exchange: "BackTestExchange", *args, **kwargs):
    if exchange.executions is None:
        logger.warning(
            "Execution transaction is disable, enable by flag: show_execution=True"
        )
        return

    super().__init__(*args, exchange=exchange, **kwargs)

is_long property

is_long: bool

True if side is long (size is positive).

Returns:

• bool ( bool ) –

  True/False

is_short property

is_short: bool

True if side is short (size is negative).

Returns:

• bool ( bool ) –

  description

side property

side: TradeSide

True if side is short (size is negative).

Returns:

• bool ( TradeSide ) –

  description

from_order classmethod

from_order(
    order: BackTestOrder,
    price: float,
    at: object,
    size: float | None = None,
) -> BackTestExecution

Method help to build Execution object from Order object

Parameters:

• price (float) –

  Executed price

• at (object) –

  Executed bar

• size (float | None, default: None ) –

  Executed size. Defaults to None.

Returns:

• BackTestExecution ( BackTestExecution ) –

  Execution object

Source code in lettrade/exchange/backtest/trade.py

@classmethod
def from_order(
    cls,
    order: "BackTestOrder",
    price: float,
    at: object,
    size: float | None = None,
) -> "BackTestExecution":
    """Method help to build Execution object from Order object

    Args:
        price (float): Executed price
        at (object): Executed bar
        size (float | None, optional): Executed size. Defaults to None.

    Returns:
        BackTestExecution: Execution object
    """
    return cls(
        id=order.id,
        size=size or order.size,
        exchange=order.exchange,
        data=order.data,
        price=price,
        at=at,
        order=order,
    )

merge

merge(other: Execution)

Merge to keep object handler but not overwrite for Strategy using when Strategy want to store object and object will be automatic update directly

Source code in lettrade/exchange/execution.py

def merge(self, other: "Execution"):
    """
    Merge to keep object handler but not overwrite for Strategy using when Strategy want to store object and object will be automatic update directly
    """
    if other is self:
        return

    if self.id != other.id:
        raise RuntimeError(f"Merge difference id {self.id} != {other.id} execution")

    self.price = other.price
    self.size = other.size

    if other.at:
        self.at = other.at
    if other.order_id:
        self.order_id = other.order_id
    if other.order:
        self.order = other.order
    if other.position_id:
        self.position_id = other.position_id
    if other.position:
        self.position = other.position

BackTestOrder

BackTestOrder(
    id: str,
    exchange: Exchange,
    data: DataFeed,
    size: float,
    state: OrderState = OrderState.Pending,
    type: OrderType = OrderType.Market,
    limit_price: float | None = None,
    stop_price: float | None = None,
    sl_price: float | None = None,
    tp_price: float | None = None,
    parent: Position | None = None,
    tag: str | None = None,
    placed_at: Timestamp | None = None,
    **kwargs
)

Bases: Order

Order for backtesting

Source code in lettrade/exchange/order.py

def __init__(
    self,
    id: str,
    exchange: "Exchange",
    data: "DataFeed",
    size: float,
    state: OrderState = OrderState.Pending,
    type: OrderType = OrderType.Market,
    limit_price: float | None = None,
    stop_price: float | None = None,
    sl_price: float | None = None,
    tp_price: float | None = None,
    parent: "Position | None" = None,
    tag: str | None = None,
    placed_at: pd.Timestamp | None = None,
    **kwargs,
):
    super().__init__(
        id=id,
        exchange=exchange,
        data=data,
        size=size,
        **kwargs,
    )

    self.type: OrderType = type
    self.state: OrderState = state
    self.limit_price: float | None = limit_price
    self.stop_price: float | None = stop_price
    self.sl_price: float | None = sl_price
    self.tp_price: float | None = tp_price
    self.parent: "Position | None" = parent
    self.tag: str | None = tag
    self.placed_at: pd.Timestamp | None = placed_at
    self.filled_at: pd.Timestamp | None = None
    self.filled_price: float | None = None

is_closed property

is_closed: bool

Flag to check Order closed

Returns:

• bool ( bool ) –

  True if state in [OrderState.Filled, OrderState.Canceled]

is_long property

is_long: bool

True if side is long (size is positive).

Returns:

• bool ( bool ) –

  True/False

is_opening property

is_opening: bool

Flag to check Order still alive

Returns:

• bool ( bool ) –

  True if state in [OrderState.Pending, OrderState.Placed, OrderState.Partial]

is_short property

is_short: bool

True if side is short (size is negative).

Returns:

• bool ( bool ) –

  description

is_sl_order property

is_sl_order: bool

Order is stop-loss order of a Position

Returns:

• bool ( bool ) –

  description

is_tp_order property

is_tp_order: bool

Order is take-profit order of a Position

Returns:

• bool ( bool ) –

  description

limit property

limit: float | None

Getter of limit_price

Returns:

• float | None –

  float | None: float or None

place_price property

place_price: float | None

Getter of place_price

Returns:

• float | None –

  float | None: float or None

side property

side: TradeSide

True if side is short (size is negative).

Returns:

• bool ( TradeSide ) –

  description

sl property

sl: float | None

Getter of sl_price

Returns:

• float | None –

  float | None: float or None

stop property

stop: float | None

Getter of stop_price

Returns:

• float | None –

  float | None: float or None

tp property

tp: float | None

Getter of tp_price

Returns:

• float | None –

  float | None: float or None

cancel

cancel(
    caller: Order | Position | None = None, **kwargs
) -> OrderResult

Cancel the Order and notify Exchange

Source code in lettrade/exchange/backtest/trade.py

def cancel(
    self,
    caller: Order | Position | None = None,
    **kwargs,
) -> "OrderResult":
    """Cancel the Order and notify Exchange"""
    if self.state != OrderState.Placed:
        raise RuntimeError(f"Order {self.id} state {self.state} is not Placed")

    if self.parent:
        if self is self.parent.sl_order:
            self.parent.sl_order = None
        elif self is self.parent.tp_order:
            self.parent.tp_order = None

    return super().cancel()

fill

fill(
    price: float, at: object, **kwargs
) -> BackTestExecution

Execution order and notify for Exchange

Parameters:

• price (float) –

  Executed price

• at (object) –

  Executed bar

Raises:

• RuntimeError –

  description

Returns:

• BackTestExecution ( BackTestExecution ) –

  Execution object

Source code in lettrade/exchange/backtest/trade.py

def fill(self, price: float, at: object, **kwargs) -> BackTestExecution:
    """Execution order and notify for Exchange

    Args:
        price (float): Executed price
        at (object): Executed bar

    Raises:
        RuntimeError: _description_

    Returns:
        BackTestExecution: Execution object
    """
    if self.state != OrderState.Placed:
        raise RuntimeError(f"Execution a {self.state} order")

    # Order
    ok = super().fill(price=price, at=at)

    # Execution is enable
    if self.exchange.executions is not None:
        execution = BackTestExecution.from_order(order=self, price=price, at=at)
        execution._on_execution()

    # Position hit SL/TP
    if self.parent:
        self.parent.exit(price=price, at=at, caller=self)
    else:
        # Position: Place and create new position
        position = BackTestPosition.from_order(order=self)

        position.entry(price=price, at=at)

    return ok

merge

merge(other: Order)

Update current Order variables by other Order

Parameters:

• other (Order) –

  Merge source Order

Raises:

• RuntimeError –

  Validate same id

Source code in lettrade/exchange/order.py

def merge(self, other: "Order"):
    """Update current `Order` variables by other `Order`

    Args:
        other (Order): Merge source `Order`

    Raises:
        RuntimeError: Validate same id
    """
    if other is self:
        return

    if self.id != other.id:
        raise RuntimeError(f"Merge difference id {self.id} != {other.id} order")

    self.size = other.size
    self.sl_price = other.sl_price
    self.tp_price = other.tp_price

    if other.limit_price:
        self.limit_price = other.limit_price
    if other.stop_price:
        self.stop_price = other.stop_price
    if other.placed_at:
        self.placed_at = other.placed_at

    if other.filled_price:
        self.filled_price = other.filled_price
    if other.filled_at:
        self.filled_at = other.filled_at

    if other.parent:
        self.parent = other.parent

place

place(
    at: Timestamp, raw: object | None = None
) -> OrderResult

Place Order Set status to OrderState.Placed. Send event to Exchange

Raises:

• RuntimeError –

  description

Returns:

• OrderResult ( OrderResult ) –

  result of Order

Source code in lettrade/exchange/order.py

def place(
    self,
    at: pd.Timestamp,
    raw: object | None = None,
) -> "OrderResult":
    """Place `Order`
    Set `status` to `OrderState.Placed`.
    Send event to `Exchange`

    Raises:
        RuntimeError: _description_

    Returns:
        OrderResult: result of `Order`
    """
    self.validate()

    if self.state != OrderState.Pending:
        raise RuntimeError(f"Order {self.id} state {self.state} is not Pending")

    self.state = OrderState.Placed
    self.placed_at = at

    logger.info("Placing new order: %s", self)

    self.exchange.on_order(self)
    return OrderResultOk(order=self, raw=raw)

update

update(
    limit_price: float | None = None,
    stop_price: float | None = None,
    sl: float | None = None,
    tp: float | None = None,
    raw: object | None = None,
) -> OrderResult

Update Order

Parameters:

• limit_price (float, default: None ) –

  description. Defaults to None.

• stop_price (float, default: None ) –

  description. Defaults to None.

• sl (float, default: None ) –

  description. Defaults to None.

• tp (float, default: None ) –

  description. Defaults to None.

Raises:

• RuntimeError –

  description

Source code in lettrade/exchange/order.py

def update(
    self,
    limit_price: float | None = None,
    stop_price: float | None = None,
    sl: float | None = None,
    tp: float | None = None,
    raw: object | None = None,
) -> "OrderResult":
    """Update Order

    Args:
        limit_price (float, optional): _description_. Defaults to None.
        stop_price (float, optional): _description_. Defaults to None.
        sl (float, optional): _description_. Defaults to None.
        tp (float, optional): _description_. Defaults to None.

    Raises:
        RuntimeError: _description_
    """
    if self.is_closed:
        raise RuntimeError(f"Update a closed order {self}")

    # TODO: validate parameters
    if limit_price is not None:
        self.limit_price = limit_price
    if stop_price is not None:
        self.stop_price = stop_price

    if sl is not None:
        self.sl_price = sl
    if tp is not None:
        self.tp_price = tp

    self.exchange.on_order(self)
    return OrderResultOk(order=self, raw=raw)

BackTestPosition

BackTestPosition(
    id: str,
    exchange: Exchange,
    data: DataFeed,
    size: float,
    parent: Order,
    state: PositionState = PositionState.Open,
    entry_price: float | None = None,
    entry_fee: float = 0.0,
    entry_at: Timestamp | None = None,
    sl_order: Order | None = None,
    tp_order: Order | None = None,
    tag: str | None = None,
    **kwargs
)

Bases: Position

Position for backtesting

Source code in lettrade/exchange/position.py

def __init__(
    self,
    id: str,
    exchange: "Exchange",
    data: "DataFeed",
    size: float,
    parent: "Order",
    state: PositionState = PositionState.Open,
    entry_price: float | None = None,
    entry_fee: float = 0.0,
    entry_at: pd.Timestamp | None = None,
    sl_order: Order | None = None,
    tp_order: Order | None = None,
    tag: str | None = None,
    **kwargs,
):
    super().__init__(
        id=id,
        exchange=exchange,
        data=data,
        size=size,
        **kwargs,
    )
    self._account: "Account" = self.exchange._account

    self.state: PositionState = state
    self.parent: "Order" = parent
    self.tag: str | None = tag

    self.entry_price: float | None = entry_price
    self.entry_fee: float = entry_fee
    self.entry_at: pd.Timestamp | None = entry_at

    self.exit_price: float | None = None
    self.exit_fee: float = 0.0
    self.exit_at: pd.Timestamp | None = None
    self.exit_pl: float | None = None

    self.sl_order: Order | None = sl_order
    self.tp_order: Order | None = tp_order

fee property

fee: float

Fee/Estimate Fee for trade

Returns:

• float ( float ) –

  Fee

is_exited property

is_exited: bool

Flag to check Position state.

Returns:

• bool ( bool ) –

  True if the trade exited

is_long property

is_long: bool

True if side is long (size is positive).

Returns:

• bool ( bool ) –

  True/False

is_opening property

is_opening: bool

Flag to check Position state.

Returns:

• bool ( bool ) –

  True if the trade opening

is_short property

is_short: bool

True if side is short (size is negative).

Returns:

• bool ( bool ) –

  description

pl property

pl: float

Estimate Profit or Loss of Position

Returns:

• float ( float ) –

  PnL

side property

side: TradeSide

True if side is short (size is negative).

Returns:

• bool ( TradeSide ) –

  description

exit

exit(
    price: float | None = None,
    at: Timestamp | None = None,
    caller: Order | Position | None = None,
    **kwargs
) -> PositionResult

Exit Position

Parameters:

• price (float, default: None ) –

  Exit price

• at (object, default: None ) –

  Exit bar

• caller (Order | Position, default: None ) –

  Skip caller to prevent infinite recursion loop. Defaults to None.

Source code in lettrade/exchange/backtest/trade.py

def exit(
    self,
    price: float | None = None,
    at: pd.Timestamp | None = None,
    caller: Order | Position | None = None,
    **kwargs,
) -> PositionResult:
    """Exit Position

    Args:
        price (float): Exit price
        at (object): Exit bar
        caller (Order | Position, optional): Skip caller to prevent infinite recursion loop. Defaults to None.
    """
    if self.state == PositionState.Exit:
        if caller is None:
            # Call by user
            raise RuntimeError(f"Call exited position {self}")
        return

    if caller is None:
        # Call by user
        if price is not None:
            raise RuntimeError(f"Price set {price} is not available")
        if at is not None:
            raise RuntimeError(f"At set {at} is not available")

        price = self.data.l.open[0]
        at = self.data.l.index[0]
    else:
        # Call by SL/TP order
        if price is None or at is None:
            raise RuntimeError(f"Caller {caller} with price is None or at is None")

    # PnL
    pl = self._account.pl(
        size=self.size,
        entry_price=self.entry_price,
        exit_price=price,
    )

    # Fee
    fee = self._account.fee(size=self.size)

    # State
    ok = super().exit(price=price, at=at, pl=pl, fee=fee, **kwargs)

    # Caller is position close by tp/sl order
    if caller is None:
        if self.sl_order is not None:
            self.sl_order.cancel(caller=self)
        if self.tp_order is not None:
            self.tp_order.cancel(caller=self)
    else:
        if self.sl_order and self.sl_order is not caller:
            self.sl_order.cancel(caller=caller)
        if self.tp_order and self.tp_order is not caller:
            self.tp_order.cancel(caller=caller)

    return ok

from_order classmethod

from_order(
    order: BackTestOrder,
    size: float | None = None,
    state: PositionState = PositionState.Open,
    **kwargs
) -> BackTestPosition

Build Position object from Order object

Parameters:

• size (float, default: None ) –

  Size of Position object. Defaults to None.

• state (PositionState, default: Open ) –

  State of Position object. Defaults to PositionState.Open.

Returns:

• BackTestPosition ( BackTestPosition ) –

  Position object

Source code in lettrade/exchange/backtest/trade.py

@classmethod
def from_order(
    cls,
    order: "BackTestOrder",
    size: float | None = None,
    state: PositionState = PositionState.Open,
    **kwargs,
) -> "BackTestPosition":
    """Build Position object from Order object

    Args:
        size (float, optional): Size of Position object. Defaults to None.
        state (PositionState, optional): State of Position object. Defaults to PositionState.Open.

    Returns:
        BackTestPosition: Position object
    """
    position = cls(
        id=order.id,
        size=size or order.size,
        exchange=order.exchange,
        data=order.data,
        state=state,
        parent=order,
    )
    if order.sl_price:
        position._new_sl_order(stop_price=order.sl_price)
    if order.tp_price:
        position._new_tp_order(limit_price=order.tp_price)
    order.parent = position
    return position

merge

merge(other: Position) -> bool

Merge position from another position has same id

Parameters:

• other (Position) –

  description

Raises:

• RuntimeError –

  description

Source code in lettrade/exchange/position.py

def merge(self, other: "Position") -> bool:
    """Merge position from another position has same id

    Args:
        other (Position): _description_

    Raises:
        RuntimeError: _description_
    """
    if other is self:
        return False

    if self.id != other.id:
        raise RuntimeError(f"Merge difference id {self.id} != {other.id} order")

    self.size = other.size

    if other.sl_order is not None:
        self.sl_order = other.sl_order
        self.sl_order.parent = self
    elif self.sl_order is not None:
        self.sl_order.cancel()
        self.sl_order = None

    if other.tp_order is not None:
        self.tp_order = other.tp_order
        self.tp_order.parent = self
    elif self.tp_order is not None:
        self.tp_order.cancel()
        self.tp_order = None

    if other.entry_at:
        self.entry_at = other.entry_at
    if other.entry_price:
        self.entry_price = other.entry_price

    if other.exit_at:
        self.exit_at = other.exit_at
    if other.exit_price:
        self.exit_price = other.exit_price
    if other.exit_fee:
        self.exit_fee = other.exit_fee
    if other.exit_pl:
        self.exit_pl = other.exit_pl

    if other.parent:
        self.parent = other.parent

    return True