Epic
Epic Class
An Epic
is a financial instrument.
Attributes:
Name | Type | Description |
---|---|---|
trade_provider |
BaseTradeProvider |
trade provider of this instance |
timezone |
str |
timezone to apply to all received ticks |
open_time |
datetime.time |
Epic open time |
close_time |
datetime.time |
current instance close time |
trade_days |
List[int] |
List of trading days for this Epic (monday=0) |
holidays |
List[datetime.date] |
List of holidays for this instance. |
frame_sets |
Dict[str, "FrameSet"] |
Dict of frame sets indexed by their reference. |
strategies |
Dict[str, "BaseStrategy"] |
Dict of strategies indexed by their reference. |
ref |
str |
reference of this instance
(see |
ref: str
inherited
property
writable
Return ref of current instance.
Returns:
Type | Description |
---|---|
str |
reference of current instance. |
timezone: str
property
writable
Return the current instance timezone.
Returns:
Type | Description |
---|---|
str |
epic timezone (see |
__init__(self, timezone='UTC', open_time=datetime.time(9, 30), close_time=datetime.time(17, 30), trade_days=None, holidays=None, trade_provider=None, ref=None)
special
Create a new Epic instance.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
trade_provider |
Optional[BaseTradeProvider] |
instance inheriting from
|
None |
timezone |
str |
timezone string from |
'UTC' |
open_time |
time |
Instance open_time |
datetime.time(9, 30) |
close_time |
time |
Instance close_time |
datetime.time(17, 30) |
trade_days |
Optional[List[int]] |
List of weekdays when the market is open (Monday=0). |
None |
holidays |
Optional[List[datetime.date]] |
List of dates where the Epic is open. |
None |
ref |
Optional[str] |
epic code name |
None |
Exceptions:
Type | Description |
---|---|
estrade.EpicException |
if timezone is invalid |
Source code in estrade/epic.py
def __init__(
self,
timezone: str = "UTC",
open_time: time = time(9, 30),
close_time: time = time(17, 30),
trade_days: Optional[List[int]] = None,
holidays: Optional[List[date]] = None,
trade_provider: Optional["BaseTradeProvider"] = None,
ref: Optional[str] = None,
) -> None:
"""
Create a new Epic instance.
Arguments:
trade_provider: instance inheriting from
[`TradeProvider`][estrade.trade_provider.BaseTradeProvider]. Defaults to
[`TradeProviderBacktests`][estrade.trade_provider.TradeProviderBacktests]
if not provided
timezone: timezone string from `pytz.all_timezones`, it is applied on tick
on association with Epic.
open_time: Instance open_time
close_time: Instance close_time
trade_days: List of weekdays when the market is open (Monday=0).
holidays: List of dates where the Epic is open.
ref: epic code name
Raises:
estrade.EpicException: if timezone is invalid
"""
# call parent mixins
RefMixin.__init__(self, ref)
# handle arguments to set attributes
self.timezone: str = timezone
self.open_time = open_time
self.close_time = close_time
self.trade_days = trade_days or [0, 1, 2, 3, 4]
self.holidays = holidays or []
self.trade_provider = trade_provider or TradeProviderBacktests()
# create a fake tick to init last_tick attribute
self.last_tick: Tick = Tick(
datetime=arrow.get("1900-01-01 00:00:00"),
bid=0,
ask=0,
)
self.frame_sets: Dict[str, "FrameSet"] = {}
self.strategies: Dict[str, "BaseStrategy"] = {}
self.market_open: bool = False
logger.info(f"New Epic created: {self}")
__str__(self)
special
Return a string representation of the current instance.
Returns:
Type | Description |
---|---|
str |
string representation of current |
Source code in estrade/epic.py
def __str__(self) -> str:
"""
Return a string representation of the current instance.
Returns:
string representation of current [`Epic`][estrade.epic.Epic] instance.
"""
return f"Epic: {self.ref} on timezone {self.timezone}"
close_trade(self, trade, **kwargs)
Close a new trade on this Epic.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
trade |
Trade |
Trade to close |
required |
kwargs |
|
see arguments of |
{} |
Source code in estrade/epic.py
def close_trade(self, trade: Trade, **kwargs) -> TradeClose:
"""
Close a new trade on this Epic.
Arguments:
trade: Trade to close
kwargs: see arguments of [`TradeClose`][estrade.trade.TradeClose]
"""
logger.debug("Close a trade for epic %s", self.ref)
trade_close = trade.close_from_epic(**kwargs)
self.trade_provider.close_trade(trade_close)
return trade_close
get_frame(self, frame_set_ref, offset=0)
Get a frame from a FrameSet.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
frame_set_ref |
str |
reference of the
|
required |
offset |
int |
number of frames backwards (eg: 0 = current frame, 1 = last closed frame) |
0 |
Returns:
Type | Description |
---|---|
Optional[Frame] |
frame |
Source code in estrade/epic.py
def get_frame(self, frame_set_ref: str, offset: int = 0) -> Optional["Frame"]:
"""
Get a frame from a FrameSet.
Arguments:
frame_set_ref: reference of the
[`FrameSet`][estrade.graph.frame_set.FrameSet]
offset: number of frames backwards
(eg: 0 = current frame, 1 = last closed frame)
Returns:
frame
"""
offset = (offset + 1) * -1
frame_set = self.frame_sets[frame_set_ref]
try:
frame = frame_set[offset]
except IndexError:
return None
return frame
get_indicator_value(self, frame_set_ref, indicator_ref, offset=0)
Get indicator value for a FrameSet.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
frame_set_ref |
str |
reference of the
|
required |
indicator_ref |
str |
indicator reference |
required |
offset |
int |
number of frames backwards (eg: 0 = current frame, 1 = last closed frame) |
0 |
Returns:
Type | Description |
---|---|
Optional[BaseIndicatorValue] |
Indicator value |
Source code in estrade/epic.py
def get_indicator_value(
self, frame_set_ref: str, indicator_ref: str, offset: int = 0
) -> Optional["BaseIndicatorValue"]:
"""
Get indicator value for a FrameSet.
Arguments:
frame_set_ref: reference of the
[`FrameSet`][estrade.graph.frame_set.FrameSet]
indicator_ref: indicator reference
offset: number of frames backwards
(eg: 0 = current frame, 1 = last closed frame)
Returns:
Indicator value
"""
frame = self.get_frame(frame_set_ref=frame_set_ref, offset=offset)
if frame is None:
return None
indicator_value = frame[indicator_ref]
return indicator_value
is_market_open(self)
Check if the market is currently opened.
This method compare the last received tick datetime to the current instance
open_time
and close_time
.
Returns:
Type | Description |
---|---|
bool |
|
Source code in estrade/epic.py
def is_market_open(self) -> bool:
"""
Check if the market is currently opened.
This method compare the last received tick datetime to the current instance
`open_time` and `close_time`.
Returns:
- `None` when the epic contains no Tick
- `True` if the last received tick datetime
- datetime is between `open_time` and `close_time`
- day is not a holiday
- weekday is a trade_days
- `False` in other cases
"""
if not self._in_market_hours(self.last_tick.datetime.time()):
logger.debug("Tick is not in Market hours")
return False
elif self.last_tick.datetime.weekday() not in self.trade_days:
logger.debug("Tick is not in a valid weekday")
return False
elif self.last_tick.datetime.date() in self.holidays:
logger.debug("Tick is not in a holiday")
return False
return True
on_new_tick(self, tick)
Add a new tick to Epic.
The purpose of this method is to:
- Apply Epic timezone to the received Tick datetime
- Update frames sets
- Update opened trades result
- Trigger strategies
Use this method in your TickProvider to propagate every new received Tick.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
tick |
Tick |
new tick |
required |
Source code in estrade/epic.py
def on_new_tick(self, tick: "Tick") -> None:
"""
Add a new tick to Epic.
The purpose of this method is to:
1. Apply Epic timezone to the received Tick datetime
3. Update frames sets
4. Update opened trades result
4. Trigger strategies
Use this method in your TickProvider to propagate every new received Tick.
Arguments:
tick: new tick
"""
# set tick timezone
tick.datetime = tick.datetime.to(self.timezone)
if tick.datetime < self.last_tick.datetime:
raise EpicException(
"Cannot handle a tick anterior to the last received tick."
)
# update last tick
self.last_tick = tick
market_open_before_new_tick = self.market_open
self.market_open = self.is_market_open()
self._update_frame_sets(tick)
self._update_open_trades(tick)
self._execute_strategies(market_open_before_new_tick)
open_trade(self, **kwargs)
Open a new trade on this Epic.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
kwargs |
|
see arguments of |
{} |
Source code in estrade/epic.py
def open_trade(self, **kwargs) -> Trade:
"""
Open a new trade on this Epic.
Arguments:
kwargs: see arguments of [`Trade`][estrade.trade.Trade]
"""
logger.debug("Open a new trade for epic %s: %s", self.ref, kwargs)
new_trade = Trade.open_from_epic(epic=self, **kwargs)
self.trade_provider.open_trade(new_trade)
return new_trade