Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_init(self):
self.assertEqual(self.execution.EXCHANGE, ExchangeType.BETFAIR)
def test_add_execution_paper(self):
self.base_client.paper_trade = True
self.base_client.EXCHANGE = ExchangeType.BETFAIR
mock_flumine = mock.Mock()
self.base_client.add_execution(mock_flumine)
self.assertEqual(self.base_client.execution, mock_flumine.simulated_execution)
class BaseOrderType:
EXCHANGE = None
ORDER_TYPE = None
def place_instruction(self) -> dict:
raise NotImplementedError
@property
def info(self):
raise NotImplementedError
class LimitOrder(BaseOrderType):
EXCHANGE = ExchangeType.BETFAIR
ORDER_TYPE = OrderTypes.LIMIT
def __init__(
self,
price: float,
size: float,
persistence_type: str = "LAPSE",
time_in_force: str = None,
min_fill_size: float = None,
bet_target_type: str = None,
bet_target_size: float = None,
):
self.price = price
self.size = size
self.persistence_type = persistence_type
self.time_in_force = time_in_force
@property
def market_version(self) -> Optional[dict]:
return None
# todo return {"version": self.market.market_book.version}
def __iter__(self) -> Iterator[BaseOrder]:
return iter(self.orders)
def __len__(self) -> int:
return len(self.orders)
class BetfairOrderPackage(BaseOrderPackage):
EXCHANGE = ExchangeType.BETFAIR
@property
def place_instructions(self):
return [order.create_place_instruction() for order in self]
@property
def cancel_instructions(self):
return [
order.create_cancel_instruction() for order in self
] # todo? if order.size_remaining > 0
@property
def update_instructions(self):
return [order.create_update_instruction() for order in self]
@property
"size_voided": self.size_voided,
"average_price_matched": self.average_price_matched,
},
"status": self.status.value if self.status else None,
"status_log": ", ".join([s.value for s in self.status_log]),
}
def __repr__(self):
return "Order {0}: {1}".format(
self.bet_id, self.status.value if self.status else None
)
class BetfairOrder(BaseOrder):
EXCHANGE = ExchangeType.BETFAIR
# updates
def place(self, publish_time: int) -> None:
self.publish_time = publish_time
self.placing()
def cancel(self, size_reduction: float = None) -> None:
if self.order_type.ORDER_TYPE == OrderTypes.LIMIT:
if size_reduction and self.size_remaining - size_reduction < 0:
raise OrderUpdateError("Size reduction too large")
if self.status != OrderStatus.EXECUTABLE:
raise OrderUpdateError("Current status: %s" % self.status)
self.update_data["size_reduction"] = size_reduction
self.cancelling()
else:
raise OrderUpdateError(
def _validate(self, order_package):
for order in order_package:
if order.EXCHANGE == ExchangeType.BETFAIR:
self._validate_betfair_order(order)
def add_execution(self, flumine) -> None:
if self.EXCHANGE == ExchangeType.SIMULATED or self.paper_trade:
self.execution = flumine.simulated_execution
elif self.EXCHANGE == ExchangeType.BETFAIR:
self.execution = flumine.betfair_execution
from .clients import ExchangeType
logger = logging.getLogger(__name__)
# default to GBP on error
MIN_BET_SIZE = 2
MIN_BSP_LIABILITY = 10
MIN_BET_PAYOUT = 10
class BetfairClient(BaseClient):
"""
Betfair betting client.
"""
EXCHANGE = ExchangeType.BETFAIR
def login(self) -> resources.LoginResource:
if self.interactive_login:
return self.betting_client.login_interactive()
else:
return self.betting_client.login()
def keep_alive(self) -> resources.KeepAliveResource:
if self.betting_client.session_expired:
return self.betting_client.keep_alive()
def logout(self) -> resources.LogoutResource:
return self.betting_client.logout()
def update_account_details(self) -> None:
# get details
import logging
import requests
from typing import Callable
from betfairlightweight import BetfairError
from .baseexecution import BaseExecution
from ..clients.clients import ExchangeType
from ..order.orderpackage import BaseOrderPackage, OrderPackageType
from ..exceptions import OrderExecutionError
logger = logging.getLogger(__name__)
class BetfairExecution(BaseExecution):
EXCHANGE = ExchangeType.BETFAIR
def execute_place(
self, order_package: BaseOrderPackage, http_session: requests.Session
) -> None:
response = self._execution_helper(self.place, order_package, http_session)
if response:
for (order, instruction_report) in zip(
order_package, response.place_instruction_reports
):
with order.trade:
self._order_logger(
order, instruction_report, OrderPackageType.PLACE
)
if instruction_report.status == "SUCCESS":
order.executable()
elif instruction_report.status == "FAILURE":