Browse Source

Merge remote-tracking branch 'origin/master' into barthpaleologue

Je met mon SMA dans un fichier SMA2
master
Barthélemy Paléologue 7 months ago
parent
commit
16f34a594b
  1. 5
      auto_trading/bot.py
  2. 7
      auto_trading/broker/backtest.py
  3. 28
      auto_trading/indicators/ema.py
  4. 45
      auto_trading/indicators/sma.py
  5. 55
      auto_trading/indicators/sma2.py
  6. 15
      auto_trading/interfaces.py
  7. 4
      auto_trading/ptf/in_memory.py
  8. 13
      main.py
  9. 6
      tests/broker/test_backtest.py
  10. 39
      tests/indicators/test_ema.py
  11. 34
      tests/indicators/test_sma.py
  12. 8
      tests/interfaces/test_ptf.py

5
auto_trading/bot.py

@ -26,10 +26,7 @@ class Bot:
def run(self):
"""run the bot"""
iterator = (
tqdm(self.broker) if self.logger.level >= logging.DEBUG else self.broker
)
for data in iterator:
for data in tqdm(self.broker):
self.run_once(data)
def run_once(self, data: DataFrame):

7
auto_trading/broker/backtest.py

@ -29,6 +29,10 @@ class Backtest(DataBroker):
def cursor(self) -> datetime.datetime:
return self.dates[self._cursor]
@property
def change_rate_history(self) -> pd.DataFrame:
return self.data.close.unstack(level=1)
@staticmethod
def _prepare_data(f_name: str, **kwargs) -> pd.DataFrame:
data = pd.read_csv(f_name, index_col=[0, 1], parse_dates=True, **kwargs)
@ -67,3 +71,6 @@ class Backtest(DataBroker):
except IndexError:
self._cursor -= 1
raise StopIteration
def __len__(self) -> int:
return len(self.dates) - self.start

28
auto_trading/indicators/ema.py

@ -0,0 +1,28 @@
"""EMA indicator."""
import pandas as pd # type: ignore
from ..interfaces import Indicator
class EMA(Indicator):
"""
https://stackoverflow.com/questions/48613151/simple-python-pandas-ema-ewma
"""
def __init__(self, nb_values: int):
"""Save the value."""
super().__init__()
self.nb_values = nb_values
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.
"""
return data.close.unstack().ewm(self.nb_values).mean().loc[data.index[-1][0]]

45
auto_trading/indicators/sma.py

@ -1,18 +1,18 @@
"""SMA indicator, not ready yet."""
from calendar import c
from re import sub
"""EMA indicator."""
import pandas as pd # type: ignore
from ..interfaces import Indicator
class SMA(Indicator):
"""Replay the value."""
"""
https://stackoverflow.com/questions/48613151/simple-python-pandas-ema-ewma
"""
def __init__(self, windowSize: int):
def __init__(self, nb_values: int):
"""Save the value."""
super().__init__()
self.windowSize = windowSize
self.nb_values = nb_values
def __call__(self, data: pd.DataFrame) -> pd.Series:
"""Return a dataframe of valuation of each stock from the input data.
@ -25,33 +25,6 @@ class SMA(Indicator):
DataFrame: Stock valuated float.
For each stock give -1 if realy bad and +1 if realy good.
"""
# only use date as index => actions become columns
data2 = data.unstack()
# select high prices for each action
data3 = data2["close"]
# lastND.rolling(window=2).mean()
sma = data3.rolling(window=self.windowSize).mean()
res = {}
for column in sma.columns:
# pour chaque type d'action
# print(column)
if len(sma.index) > 1:
# print(last2D[column].get(last2D.index[-1]), last2D[column].get(last2D.index[-2]))
ultieme = sma[column].get(sma.index[-1])
penultieme = sma[column].get(sma.index[-2])
if ultieme / penultieme > 1.05:
res[column] = 1
elif ultieme / penultieme < 0.95:
res[column] = -1
else:
res[column] = 0
else:
res[column] = 1
return pd.Series(res)
return (
data.close.unstack().rolling(self.nb_values).mean().loc[data.index[-1][0]]
)

55
auto_trading/indicators/sma2.py

@ -0,0 +1,55 @@
"""SMA indicator, not ready yet."""
from calendar import c
from re import sub
import pandas as pd # type: ignore
from ..interfaces import Indicator
class SMA(Indicator):
"""Replay the value."""
def __init__(self, windowSize: int):
"""Save the value."""
super().__init__()
self.windowSize = windowSize
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.
"""
# only use date as index => actions become columns
data2 = data.unstack()
# select high prices for each action
data3 = data2["close"]
# lastND.rolling(window=2).mean()
sma = data3.rolling(window=self.windowSize).mean()
res = {}
for column in sma.columns:
# pour chaque type d'action
# print(column)
if len(sma.index) > 1:
# print(last2D[column].get(last2D.index[-1]), last2D[column].get(last2D.index[-2]))
ultieme = sma[column].get(sma.index[-1])
penultieme = sma[column].get(sma.index[-2])
if ultieme / penultieme > 1.05:
res[column] = 1
elif ultieme / penultieme < 0.95:
res[column] = -1
else:
res[column] = 0
else:
res[column] = 1
return pd.Series(res)

15
auto_trading/interfaces.py

@ -69,6 +69,10 @@ class DataBroker(ABC):
For each time and each stock give (high, low, open, close).
"""
@abstractmethod
def __len__(self) -> int:
"""Total number of values"""
class Indicator(ABC):
"""Somethink that give you an insight of the market."""
@ -138,17 +142,19 @@ class PTF(ABC):
"""Somethink that buy or sell stocks."""
executors: Dict[Any, Callable[[Any, Any], None]] = {}
history: List[Order]
orders_history: List[Order]
states_history: List[PTFState]
def __init__(self, skip_errors: bool = True, save_errors: bool = True):
"""Init the class.
Args:
skip_errors (bool, optional): Do we skip orders in failure ? Defaults to True.
save_errors (bool, optional): Do we save orders in failure in the history ? Defaults to True.
save_errors (bool, optional): Do we save orders in failure in the orders_history ? Defaults to True.
"""
self.logger = logging.getLogger(self.__class__.__name__)
self.history = []
self.orders_history = []
self.states_history = []
self.skip_errors = skip_errors
self.save_errors = save_errors
@ -182,7 +188,8 @@ class PTF(ABC):
self.logger.warning("Got an order exception : %s", e.message)
order.successfull = False
if self.save_errors or order.successfull:
self.history.append(order)
self.orders_history.append(order)
self.states_history.append(self.state)
def _execute(self, order: Order) -> None:
"""Execute one order.

4
auto_trading/ptf/in_memory.py

@ -1,4 +1,4 @@
from typing import Dict, Callable
from typing import Dict, Callable, List
from ..interfaces import PTF, PTFState
from ..orders import Long, Short
@ -28,7 +28,7 @@ class InMemoryPortfolio(PTF):
return PTFState(self._balance, self._stocks.copy())
@property
def change_rate(self):
def change_rate(self) -> Dict[str, float]:
return self.change_rate_getter()
def execute_short(self, order: Short) -> None:

13
main.py

@ -1,4 +1,6 @@
import matplotlib.pyplot as plt
import pandas as pd # type: ignore
import logging
from auto_trading.broker.backtest import Backtest
from auto_trading.indicators.dumb import Dumb
@ -11,6 +13,10 @@ from auto_trading.ptf.in_memory import InMemoryPortfolio
from auto_trading.bot import Bot
pd.options.plotting.backend = "plotly"
logging.basicConfig(level=logging.DEBUG)
if __name__ == "__main__":
bt = Backtest("./data/NYSE_small.csv")
ptf = InMemoryPortfolio(
@ -44,3 +50,10 @@ if __name__ == "__main__":
print(bot.ptf.total_balance(bot.broker.current_change))
plt.show()
ch_history: pd.DataFrame = bot.broker.change_rate_history # type: ignore
st_history = pd.DataFrame(
[s.stocks for s in bot.ptf.states_history],
index=ch_history.index,
)
(st_history * ch_history).fillna(0).plot.area().show()

6
tests/broker/test_backtest.py

@ -1,6 +1,7 @@
import pytest
import datetime
from pandas import DataFrame # type: ignore
from auto_trading.broker.backtest import Backtest
@ -22,6 +23,11 @@ def test_nyse_backtest(backtest, first_change):
assert backtest.current_change == first_change
assert backtest.properties.period == step
assert backtest.step == step
assert (
(backtest.change_rate_history == backtest.data.close.unstack(level=1))
.all()
.all()
)
k = first_day
for data in backtest:
k += step

39
tests/indicators/test_ema.py

@ -0,0 +1,39 @@
"""Realy basic tests."""
import pytest
from datetime import datetime
from pandas import DataFrame, Series # type: ignore
from auto_trading.indicators.ema import EMA
def date(j: int) -> datetime:
return datetime.strptime(f"2015-03-{j}", "%Y-%m-%d")
@pytest.mark.parametrize(
"nb_values, data, expected",
[
(
10,
DataFrame(
{
"close": {
(date(i), st): 10 if i == 10 and st == "GOOG" else 0
for i in range(1, 11)
for st in ("GOOG", "GOOGL")
}
}
),
{"GOOG": 1.48, "GOOGL": 0.0},
),
],
)
def test_ema(nb_values: int, data: DataFrame, expected: Series) -> None:
"""Test the ema indicator."""
ema = EMA(nb_values)
res = ema(data).to_dict()
# round all values
res = {k: round(v, 3) for k, v in res.items()}
assert res == expected

34
tests/indicators/test_sma.py

@ -0,0 +1,34 @@
"""Realy basic tests."""
import pytest
from datetime import datetime
from pandas import DataFrame, Series # type: ignore
from auto_trading.indicators.sma import SMA
def date(j: int) -> datetime:
return datetime.strptime(f"2015-03-{j}", "%Y-%m-%d")
@pytest.mark.parametrize(
"nb_values, data, expected",
[
(
10,
DataFrame(
{
"close": {
(date(i), st): 10 if i == 10 and st == "GOOG" else 0
for i in range(1, 11)
for st in ("GOOG", "GOOGL")
}
}
),
{"GOOG": 1.0, "GOOGL": 0.0},
),
],
)
def test_sma(nb_values: int, data: DataFrame, expected: Series) -> None:
"""Test the SMA indicator."""
sma = SMA(nb_values)
assert sma(data).to_dict() == expected

8
tests/interfaces/test_ptf.py

@ -13,7 +13,7 @@ short_order = Short("GOLD", 1, 1)
class _TestPTFLong(PTF):
@property
def state(self):
return PTFState(0, {}, DataFrame())
return PTFState(0, {})
def execute_long(self, order: Long) -> None:
assert order == long_order
@ -24,7 +24,7 @@ class _TestPTFLong(PTF):
class _TestPTFLongShort(PTF):
@property
def state(self):
return PTFState(0, {}, DataFrame())
return PTFState(0, {})
def execute_long(self, order: Long) -> None:
assert order == long_order
@ -38,7 +38,7 @@ class _TestPTFLongShort(PTF):
class _TestPTFShort(PTF):
@property
def state(self):
return PTFState(0, {}, DataFrame())
return PTFState(0, {})
def execute_short(self, order: Long) -> None:
assert order == short_order
@ -49,7 +49,7 @@ class _TestPTFShort(PTF):
class _TestPTFNone(PTF):
@property
def state(self):
return PTFState(0, {}, DataFrame())
return PTFState(0, {})
@pytest.mark.parametrize(

Loading…
Cancel
Save