Format

Makefile

@@ -1,3 +1,5 @@
 doc:
 	pdoc src/pybottrader --output-dir docs
 
+format:
+	black .

src/pybottrader/datastreamers.py

@@ -5,32 +5,41 @@
 import pandas as pd
 from .strategies import Position
 
+
 @define
 class StreamIteration:
     """Used to report results from a stream iteration"""
+
     time: pd.Timestamp
     position: Position
     data: dict
     roi: Union[float, None]
     portfolio_value: float
     accumulated_roi: Union[float, None]
 
+
 class DataStreamer:
     """A data streamer abstract class"""
+
     def __init__(self):
         """Init method"""
+
     def next(self) -> Union[dict, None]:
         """Next method"""
 
+
 class CSVFileStreamer(DataStreamer):
     """
     An dataframe file streamer
     """
+
     data: pd.DataFrame
     index: int
+
     def __init__(self, filename: str):
         self.index = 0
         self.data = pd.read_csv(filename, parse_dates=True)
+
     def next(self) -> Union[dict, None]:
         if self.index >= len(self.data):
             return None

src/pybottrader/indicators.py

@@ -41,17 +41,21 @@ def get(self) -> float:
             return np.nan
         return self.accum / self.period
 
+
 class EMA:
     """Exponential Moving Average"""
+
     periods: float
     alpha: float
     smooth_factor: float
     length: int = 0
     prev: float = 0.0
-    def __init__(self, periods: int, alpha = 2.0):
+
+    def __init__(self, periods: int, alpha=2.0):
         self.periods = periods
         self.alpha = alpha
         self.smooth_factor = alpha / (1.0 + periods)
+
     def update(self, value: float) -> float:
         """Aggregate a new value into the moving average"""
         self.length += 1
@@ -61,16 +65,18 @@ def update(self, value: float) -> float:
             self.prev += value
             self.prev /= self.periods
         else:
-            self.prev = (value * self.smooth_factor) + self.prev * (1.0 - self.smooth_factor)
+            self.prev = (value * self.smooth_factor) + self.prev * (
+                1.0 - self.smooth_factor
+            )
         return self.get()
 
-
     def get(self) -> float:
         """Current output of the moving average"""
         if self.length < self.periods:
             return np.nan
         return self.prev
 
+
 def roi(initial_value, final_value):
     """Return on investment"""
     if initial_value == 0:

src/pybottrader/portfolios.py

@@ -8,55 +8,59 @@
 from .strategies import Position
 from .indicators import roi
 
+
 class Portfolio:
     """Base Portfolio Class"""
+
     initial_cash: float
     last_position: Position
     last_price: float
     last_ticker: str
+
     def __init__(self, cash: float = 1000.0):
         """Init method"""
         self.initial_cash = cash
         self.last_position = Position.STAY
         self.last_price = 0.0
         self.last_ticker = ""
+
     def process(
-        self,
-        ticker: str = "",
-        position: Position = Position.STAY,
-        price: float = 0.0
+        self, ticker: str = "", position: Position = Position.STAY, price: float = 0.0
     ):
         """Process signal"""
         self.last_ticker = ticker
         self.last_price = price
         self.last_position = position
+
     def valuation(self) -> float:
         """Default valuation method"""
         return self.initial_cash
+
     def accumulated_return(self) -> float:
         """Accumulated ROI"""
         return roi(self.initial_cash, self.valuation())
 
+
 class DummyPortfolio(Portfolio):
     """
     Dummy portfolio is the most basic portfolio model.
     It works with only one asset. When it receives the buy signal,
     it uses all the available cash to buy the asset. When it receives
     the sell signal, it sells all the shares of the asset.
     """
+
     cash: float
     share_units: float
     share_price: float
+
     def __init__(self, cash: float = 1000.0):
         super().__init__(cash)
         self.cash = cash
         self.share_units = 0.0
         self.share_price = 0.0
+
     def process(
-        self,
-        ticker: str = "",
-        position: Position = Position.STAY,
-        price: float = 0.0
+        self, ticker: str = "", position: Position = Position.STAY, price: float = 0.0
     ):
         super().process(ticker=ticker, position=position, price=price)
         if position == Position.BUY:
@@ -71,5 +75,6 @@ def process(
             self.cash = self.share_units * price
             self.share_price = price
             self.share_units = 0.0
+
     def valuation(self) -> float:
         return self.cash if self.cash > 0.0 else (self.share_price * self.share_units)

src/pybottrader/strategies.py

@@ -2,6 +2,7 @@
 
 from enum import Enum
 
+
 class Position(Enum):
     """Trading Positions"""
 
@@ -24,6 +25,3 @@ def evaluate(self, *args, **kwargs) -> Position:
         """
         # The default position is STAY
         return Position.STAY
-
-
-

src/pybottrader/traders.py

@@ -6,13 +6,16 @@
 from .strategies import Strategy, Position
 from .indicators import roi
 
+
 class Trader:
     """Base class"""
+
     portfolio: Portfolio
     datastream: DataStreamer
     strategy: Strategy
-    last_result : Union[StreamIteration, None] = None
+    last_result: Union[StreamIteration, None] = None
     last_valuation: float = 0.0
+
     def __init__(
         self,
         strategy: Strategy,
@@ -23,34 +26,37 @@ def __init__(
         self.datastream = datastream
         self.portfolio = portfolio
         self.strategy = strategy
+
     def next(self) -> bool:
         obs = self.datastream.next()
         if obs is None:
             return False
         pos = self.strategy.evaluate(**obs)
-        self.portfolio.process(position=pos, price=obs['close'])
+        self.portfolio.process(position=pos, price=obs["close"])
         self.last_result = StreamIteration(
-            time=obs['time'],
+            time=obs["time"],
             position=pos,
             data=obs,
             roi=roi(self.last_valuation, self.portfolio.valuation()),
             portfolio_value=self.portfolio.valuation(),
-            accumulated_roi=self.portfolio.accumulated_return()
+            accumulated_roi=self.portfolio.accumulated_return(),
         )
         self.last_valuation = self.portfolio.valuation()
         return True
+
     def status(self) -> StreamIteration:
         """Trader last result"""
         return self.last_result
+
     def run(self):
         while self.next():
             status = self.status()
             # Printing status after BUY or SELL
             if status.position != Position.STAY:
                 print(
-                    f"{status.time}  " + 
-                    f"{status.position.name:5}  " + 
-                    f"{status.data['close']:10.2f} USD  " +
-                    f"ROI {status.roi * 100.0:5.1f} %  "
+                    f"{status.time}  "
+                    + f"{status.position.name:5}  "
+                    + f"{status.data['close']:10.2f} USD  "
+                    + f"ROI {status.roi * 100.0:5.1f} %  "
                     f"Accum. ROI {status.accumulated_roi * 100.0:5.1f} %"
                 )

test/test_indicators.py

@@ -18,6 +18,7 @@ def test_ma():
         else:
             assert y == pytest.approx(ts[i] - 1.0)
 
+
 def test_ema():
     """
     This test has been adapted from:
@@ -34,6 +35,7 @@ def test_ema():
         else:
             assert abs(y - res[i - periods + 1]) < 1e-6
 
+
 def test_roi():
     assert np.isnan(roi(0, 100))
     assert abs(roi(100, 120) - 0.2) < 1e-6

test/test_portfolios.py

@@ -1,38 +1,46 @@
 from pybottrader.portfolios import *
 from pybottrader.strategies import Position
 
+
 class TestBasePortfolio:
     def setup_method(self):
         self.portfolio = Portfolio(cash=1000.0)
+
     def test_constructor(self):
         assert abs(self.portfolio.initial_cash - 1000.0) < 1e-6
         assert self.portfolio.last_position == Position.STAY
         assert self.portfolio.last_price < 1e-6
         assert self.portfolio.last_ticker == ""
+
     def test_process(self):
         self.portfolio.process(ticker="AAPL", position=Position.BUY, price=100.0)
         assert self.portfolio.last_ticker == "AAPL"
         assert self.portfolio.last_position == Position.BUY
         assert abs(self.portfolio.last_price - 100.0) < 1e-6
+
     def test_valuation(self):
         assert abs(self.portfolio.valuation() - 1000.0) < 1e-6
         self.portfolio.process(ticker="AAPL", position=Position.BUY, price=100.0)
         assert abs(self.portfolio.valuation() - 1000.0) < 1e-6
+
     def test_accumulated_return(self):
         assert self.portfolio.accumulated_return() < 1e-6
         self.portfolio.process(ticker="AAPL", position=Position.BUY, price=100.0)
         assert self.portfolio.accumulated_return() < 1e-6
 
+
 class TestDummyPortafolio:
     def setup_method(self):
         self.portfolio = DummyPortfolio(cash=1000.0)
+
     def test_constructor(self):
         assert abs(self.portfolio.cash - 1000.0) < 1e-6
         assert self.portfolio.share_units < 1e-6
         assert abs(self.portfolio.initial_cash - 1000.0) < 1e-6
         assert self.portfolio.last_position == Position.STAY
         assert abs(self.portfolio.last_price) < 1e-6
         assert self.portfolio.last_ticker == ""
+
     def test_buy(self):
         self.portfolio.process(position=Position.BUY, price=10.0)
         assert self.portfolio.last_position == Position.BUY
@@ -43,6 +51,7 @@ def test_buy(self):
         self.portfolio.process(position=Position.BUY, price=10.0)
         assert self.portfolio.cash < 1e-6
         assert abs(self.portfolio.share_units - 100.0) < 1e-6
+
     def test_sell(self):
         # Trying to sell without having any share
         self.portfolio.process(position=Position.SELL, price=10.0)
@@ -58,16 +67,17 @@ def test_sell(self):
         self.portfolio.process(position=Position.SELL, price=12.0)
         assert abs(self.portfolio.cash - 1200.0) < 1e-6
         assert self.portfolio.share_units < 1e-6
+
     def test_valuation(self):
         assert abs(self.portfolio.valuation() - 1000.0) < 1e-6
         self.portfolio.process(position=Position.BUY, price=10.0)
         assert abs(self.portfolio.valuation() - 1000.0) < 1e-6
         self.portfolio.process(position=Position.SELL, price=12.0)
         assert abs(self.portfolio.valuation() - 1200.0) < 1e-6
+
     def test_accumulated_return(self):
         assert self.portfolio.accumulated_return() < 1e-6
         self.portfolio.process(position=Position.BUY, price=10.0)
         assert self.portfolio.accumulated_return() < 1e-6
         self.portfolio.process(position=Position.SELL, price=12.0)
         assert abs(self.portfolio.accumulated_return() - 0.2) < 1e-6
-