You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

74 lines
2.7 KiB

from typing import List, Dict, Union
import pandas as pd # type: ignore
from ..orders import Long, Short
from ..interfaces import Strategy, Order, PTFState, Indicator
from ..errors import StrategyInitError
class Prop(Strategy):
"""Proportional repartion from the indicator."""
def __init__(self, indicators: Dict[str, Indicator] = None, max_delta: float = 10):
"""Init the class.
Only have one indicator.
The max delta is the maximum difference between the actual and the desired state.
if not indicators or len(indicators) != 1:
raise StrategyInitError(
"This strat must be initialized with one indicator."
self.indicator = list(indicators.keys())[0]
self.max_delta = max_delta
def filter_order(self, order: Union[Long, Short]) -> bool:
"""Return if the order value is high enough."""
return order.amount_usd > self.max_delta
def execute(
self, data: pd.DataFrame, indicators_results: pd.DataFrame, state: PTFState
) -> List[Order]:
"""Just hold the value [to_hold].
data (DataFrame): The Data broker output.
For each time and each stock give (high, low, open, close).
indicators_results (DataFrame): Indicator-Stock valuated float.
For each indicator and each stock give -1 if realy bad and +1 if realy good.
List[Order]: A list of orders to execute.
orders: List[Union[Long, Short]] = []
conversion_rate = data.loc[data.index[-1][0]].close.to_dict()
# Desired state
results = indicators_results[self.indicator]
desired_state_precent = results[results > 0] / results[results > 0].sum()
desired_state_dolards = (
state.total_balance(conversion_rate) * desired_state_precent
# Remove all action that the indicator place as negative.
for stock in results[results < 0].index:
orders.append(Short(stock, state.stocks[stock], conversion_rate[stock]))
# Create the new buy orders from with the delta between the actual and the desired state.
for stock, amount in desired_state_dolards.items():
if amount > state.stocks[stock]:
Long(stock, amount - state.stocks[stock], conversion_rate[stock])
Short(stock, state.stocks[stock] - amount, conversion_rate[stock])
# Filter low orders
orders = list(filter(self.filter_order, orders))
return orders # type: ignore