Charting tool

.env.sample

@@ -1,3 +1,4 @@
 # Rename this file as .env and add the required information
 
-FMP_TOKEN=INSERT_YOUR_TOKEN_HERE
+FMP_TOKEN=INSERT_YOUR_TOKEN_HERE
+OUTPUT_FOLDER=INSERT_YOUR_FOLDER_HERE

src/bullets/data_source/data_source_interface.py

@@ -1,3 +1,4 @@
+import ssl
 from datetime import datetime
 from abc import abstractmethod
 import aiohttp
@@ -41,7 +42,7 @@ def request(url: str, method: str = "GET", body=None) -> str:
         async def fetch() -> str:
             if url:
                 async with aiohttp.ClientSession() as session:
-                    async with session.request(url=url, method=method, json=body) as response:
+                    async with session.request(url=url, method=method, json=body, ssl=ssl.SSLContext()) as response:
                         if response.status == 200:
                             return await response.text()
 

src/bullets/portfolio/transaction.py

@@ -16,6 +16,10 @@ def __init__(self, symbol: str, nb_shares: float, theoretical_price: float, simu
         self.nb_shares = nb_shares
         self.theoretical_price = theoretical_price
         self.simulated_price = simulated_price
+        if simulated_price is None or nb_shares is None:
+            self.total_price = None
+        else:
+            self.total_price = nb_shares * simulated_price
         self.timestamp = timestamp
         self.cash_balance = cash_balance
         self.status = status

src/bullets/runner.py

@@ -1,3 +1,5 @@
+import csv
+import os
 from datetime import datetime, timedelta
 from bullets.portfolio.transaction import Status
 from bullets.strategy import Strategy
@@ -26,7 +28,8 @@ def start(self):
             self.strategy.on_resolution()
             self.strategy.portfolio.on_resolution()
         self.strategy.on_finish()
-        self._post_backtest_log()
+        logger.info("=========== Backtest complete ===========")
+        self._save_backtest_log()
 
     def _get_moments(self, resolution: Resolution, start_time: datetime, end_time: datetime):
         moments = []
@@ -60,15 +63,33 @@ def _is_market_open(time: datetime, resolution: Resolution) -> bool:
 
         return True
 
-    def _post_backtest_log(self):
-        self._update_final_timestamp()
-        logger.info("=========== Backtest complete ===========")
-        logger.info("Initial Cash : " + str(self.strategy.starting_balance))
-        logger.info("Final Balance : " + str(self.strategy.portfolio.update_and_get_balance()))
-        logger.info("Final Cash : " + str(self.strategy.portfolio.cash_balance))
-        logger.info("Profit : " + str(self.strategy.portfolio.get_percentage_profit()) + "%")
-        if isinstance(self.strategy.data_source, FmpDataSource):
-            logger.info("Remaining FMP Calls :  " + str(self.strategy.data_source.get_remaining_calls()))
+    def _save_backtest_log(self):
+        new_dir = self.strategy.output_folder + datetime.now().strftime("%Y-%m-%d %H-%M-%S")
+        os.mkdir(new_dir)
+        self._save_transactions_to_csv(new_dir + "/Transactions.csv")
+        self._save_stats_to_csv(new_dir + "/Stats.csv")
+
+    def _save_transactions_to_csv(self, file: str):
+        with open(file, 'w', newline='', encoding='utf-8') as outputFile:
+            writer = csv.writer(outputFile, delimiter=';')
+            headers = ['Status', 'Order Type', 'Time', 'Symbol', 'Share Count', 'Simulated Price', 'Total Price',
+                       'Cash Balance']
+            writer.writerow(headers)
+            for tr in self.strategy.portfolio.transactions:
+                writer.writerow([tr.status.value, tr.order_type, tr.timestamp, tr.symbol, tr.nb_shares,
+                                 tr.simulated_price, tr.total_price, tr.cash_balance])
+        logger.info("Transactions sheet saved to : " + file)
+
+    def _save_stats_to_csv(self, file: str):
+        with open(file, 'w', newline='', encoding='utf-8') as outputFile:
+            writer = csv.writer(outputFile, delimiter=';')
+            writer.writerow(["Initial Cash", self.strategy.starting_balance])
+            writer.writerow(["Final Balance", self.strategy.portfolio.update_and_get_balance()])
+            writer.writerow(["Final Cash", self.strategy.portfolio.cash_balance])
+            writer.writerow(["Profit", str(self.strategy.portfolio.get_percentage_profit()) + "%"])
+            if isinstance(self.strategy.data_source, FmpDataSource):
+                writer.writerow(["Remaining FMP Calls", self.strategy.data_source.get_remaining_calls()])
+        logger.info("Stats sheet saved to : " + file)
 
     def _update_final_timestamp(self):
         final_timestamp = self.strategy.start_time

src/bullets/strategy.py

@@ -7,12 +7,14 @@
 
 class Strategy:
     def __init__(self, resolution: Resolution, start_time: datetime, end_time: datetime, starting_balance: float,
-                 data_source: DataSourceInterface, slippage_percent: int = 25, transaction_fees: int = 1):
+                 data_source: DataSourceInterface, output_folder: str, slippage_percent: int = 25,
+                 transaction_fees: int = 1):
         self.resolution = resolution
         self.start_time = start_time
         self.end_time = end_time
         self.starting_balance = starting_balance
         self.data_source = data_source
+        self.output_folder = output_folder
         self.slippage_percent = slippage_percent
         self.transaction_fees = transaction_fees
         self.timestamp = None
@@ -73,3 +75,6 @@ def _validate_start_data(self):
 
         if self.transaction_fees is None or self.transaction_fees < 0:
             raise ValueError("Transaction fees should be positive or 0")
+
+        if self.output_folder is None or self.output_folder == "":
+            raise ValueError("Invalid strategy output folder")

src/test/test_strategy.py

@@ -13,13 +13,15 @@ class TestPortfolio(unittest.TestCase):
     END_TIME = datetime(2019, 4, 22)
     STARTING_BALANCE = 5000
     FMP_TOKEN = os.getenv("FMP_TOKEN")
+    OUTPUT_FOLDER = os.getenv("OUTPUT_FOLDER")
 
     def test_strategy(self):
         strategy = TestStrategy(resolution=self.RESOLUTION,
                                 start_time=self.START_TIME,
                                 end_time=self.END_TIME,
                                 starting_balance=self.STARTING_BALANCE,
-                                data_source=FmpDataSource(self.FMP_TOKEN, self.RESOLUTION))
+                                data_source=FmpDataSource(self.FMP_TOKEN, self.RESOLUTION),
+                                output_folder=self.OUTPUT_FOLDER)
         runner = Runner(strategy)
         runner.start()
         self.assertEqual(5000, strategy.portfolio.start_balance)
@@ -31,26 +33,32 @@ def test_strategy(self):
     def test_strategy_none_date(self):
         self.assertRaisesRegex(TypeError, "Invalid strategy date type", TestStrategy,
                                self.RESOLUTION, None, None, self.STARTING_BALANCE,
-                               FmpDataSource(self.FMP_TOKEN, self.RESOLUTION))
+                               FmpDataSource(self.FMP_TOKEN, self.RESOLUTION), self.OUTPUT_FOLDER)
 
     def test_strategy_invalid_date(self):
         self.assertRaisesRegex(ValueError, "Strategy start time has to be before end time", TestStrategy,
                                self.RESOLUTION, self.END_TIME, self.START_TIME, self.STARTING_BALANCE,
-                               FmpDataSource(self.FMP_TOKEN, self.RESOLUTION))
+                               FmpDataSource(self.FMP_TOKEN, self.RESOLUTION), self.OUTPUT_FOLDER)
 
     def test_strategy_none_resolution(self):
         self.assertRaisesRegex(TypeError, "Invalid strategy resolution type", TestStrategy,
                                None, self.START_TIME, self.END_TIME, self.STARTING_BALANCE,
-                               FmpDataSource(self.FMP_TOKEN, self.RESOLUTION))
+                               FmpDataSource(self.FMP_TOKEN, self.RESOLUTION), self.OUTPUT_FOLDER)
 
     def test_strategy_none_data_source(self):
         self.assertRaisesRegex(TypeError, "Invalid strategy data source type", TestStrategy,
-                               self.RESOLUTION, self.START_TIME, self.END_TIME, self.STARTING_BALANCE, None)
+                               self.RESOLUTION, self.START_TIME, self.END_TIME, self.STARTING_BALANCE, None,
+                               self.OUTPUT_FOLDER)
 
     def test_strategy_invalid_balance(self):
         self.assertRaisesRegex(ValueError, "Strategy starting balance should be positive", TestStrategy,
                                self.RESOLUTION, self.START_TIME, self.END_TIME, self.STARTING_BALANCE * -1,
-                               FmpDataSource(self.FMP_TOKEN, self.RESOLUTION))
+                               FmpDataSource(self.FMP_TOKEN, self.RESOLUTION), self.OUTPUT_FOLDER)
+
+    def test_strategy_none_output_folder(self):
+        self.assertRaisesRegex(ValueError, "Invalid strategy output folder", TestStrategy,
+                               self.RESOLUTION, self.START_TIME, self.END_TIME, self.STARTING_BALANCE,
+                               FmpDataSource(self.FMP_TOKEN, self.RESOLUTION), None)
 
     def test_income_statements(self):
         datasource = FmpDataSource(self.FMP_TOKEN, self.RESOLUTION)