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.

91 lines
3.4 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()
results = indicators_results.T[self.indicator]
stock_to_sell = []
# Remove all action that the indicator place as negative.
for stock in results[results < 0].index:
order = Short(stock, state.stocks[stock], conversion_rate[stock])
if self.filter_order(order):
total_money = state.balance + sum(
conversion_rate[stock_name] * amount
for stock_name, amount in state.stocks.items()
if results[stock] > 0 or stock_name in stock_to_sell
# Desired state
desired_state_precent = results[results > 0] / results[results > 0].sum()
desired_state_dolards = total_money * desired_state_precent
# 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 stock in state.stocks:
if amount > state.stocks[stock]:
amount / conversion_rate[stock] - state.stocks[stock],
state.stocks[stock] - amount / conversion_rate[stock],
# Filter low orders
orders = list(filter(self.filter_order, orders))
return orders # type: ignore