Browse Source
- La strat c'est simple : si ça monte sur les deux derniers jours, on achète sinon on vend - L'indicateur calcule cette "dérivée" naive - Beaucoup d'argentthibaudlabat-spread
5 changed files with 136 additions and 5 deletions
@ -0,0 +1,65 @@ |
|||
"""Dumb indicator, for testing purposes.""" |
|||
from calendar import c |
|||
from re import sub |
|||
import pandas as pd # type: ignore |
|||
|
|||
from ..interfaces import Indicator |
|||
|
|||
|
|||
class Slopy(Indicator): |
|||
"""Replay the value.""" |
|||
|
|||
def __init__(self): |
|||
"""Save the value.""" |
|||
super().__init__() |
|||
|
|||
def __call__(self, data: pd.DataFrame) -> pd.Series: |
|||
"""Return a dataframe of valuation of each stock from the input data. |
|||
|
|||
Args: |
|||
data (DataFrame): Time-Stock valuated candlestick data. |
|||
For each time and each stock give (high, low, open, close). |
|||
|
|||
Returns: |
|||
DataFrame: Stock valuated float. |
|||
For each stock give -1 if realy bad and +1 if realy good. |
|||
""" |
|||
|
|||
#print(data.tail()) |
|||
#print(data.index.names) |
|||
|
|||
#print(data["date"]) |
|||
|
|||
#subdata = data["high"] |
|||
|
|||
# only use date as index => actions become columns |
|||
data2 = data.unstack() |
|||
|
|||
# select high prices for each action |
|||
highData = data2["high"] |
|||
|
|||
#print(highData.tail()) |
|||
|
|||
#print(highData.last("2D")) |
|||
|
|||
# select only last two days |
|||
last2D = highData.last("2D") |
|||
|
|||
res = {} |
|||
for column in last2D.columns: |
|||
#pour chaque type d'action |
|||
#print(column) |
|||
if len(last2D.index) > 1: |
|||
#print(last2D[column].get(last2D.index[-1]), last2D[column].get(last2D.index[-2])) |
|||
|
|||
ultieme = last2D[column].get(last2D.index[-1]) |
|||
penultieme = last2D[column].get(last2D.index[-2]) |
|||
|
|||
if ultieme > penultieme: |
|||
res[column] = 1 |
|||
else: |
|||
res[column] = -1 |
|||
else: |
|||
res[column] = 1 |
|||
|
|||
return pd.Series(res) |
@ -0,0 +1,59 @@ |
|||
from typing import Dict, List |
|||
|
|||
import pandas as pd # type: ignore |
|||
|
|||
from ..orders import Long, Short |
|||
from ..interfaces import Indicator, Strategy, Order, PTFState |
|||
|
|||
|
|||
class BuyUpSellDown(Strategy): |
|||
"""Just hold some stock.""" |
|||
|
|||
def __init__(self, to_hold: str, indicators: Dict[str, Indicator]): |
|||
"""Init the class""" |
|||
super().__init__(indicators=indicators) |
|||
self.to_hold = to_hold |
|||
|
|||
def execute( |
|||
self, data: pd.DataFrame, indicators_results: pd.DataFrame, state: PTFState |
|||
) -> List[Order]: |
|||
"""Just hold the value [to_hold]. |
|||
|
|||
Args: |
|||
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. |
|||
|
|||
Returns: |
|||
List[Order]: A list of orders to execute. |
|||
""" |
|||
|
|||
nb_stock_type = len(indicators_results.columns) |
|||
|
|||
orders = [] |
|||
# if I have some money |
|||
for stock_name in indicators_results.columns: |
|||
#for each stock |
|||
if (balance := state.balance) > 0: |
|||
# I calculate the value of the stock |
|||
market_price = data.loc[data.index[-1][0]].close.to_dict().get(stock_name) |
|||
|
|||
trust = int(indicators_results[stock_name]) |
|||
print(trust) |
|||
|
|||
if market_price is None: |
|||
# retry later |
|||
return [] |
|||
|
|||
amount = balance / (market_price * nb_stock_type) |
|||
|
|||
if trust > 0: |
|||
# and I buy it all at market price |
|||
orders.append(Long(stock=stock_name, amount=amount, price=market_price)) |
|||
elif trust < 0: |
|||
# and I sell it all at market price |
|||
orders.append(Short(stock=stock_name, amount=amount, price=market_price)) |
|||
|
|||
return orders # type: ignore |
Loading…
Reference in new issue