Hermes (Plutus) integration to the project.

.gitignore

@@ -48,3 +48,9 @@ coverage.xml
 
 # Sphinx documentation
 docs/_build/
+
+*.zip
+runs/*
+.idea/*
+configs/usr/db_account.yaml
+configs/usr/redis_account.yaml

requirements.txt

@@ -17,4 +17,6 @@ six==1.16.0
 tenacity==8.2.3
 tzdata==2023.3
 vnstock==0.2.8.7
-kaleido
+kaleido
+optuna
+plutus --index-url=http://localhost:3141/testuser/devs

src/bot/bot.py → src/deepmm/bot/bot.py

@@ -4,13 +4,13 @@
 from tqdm import tqdm
 from copy import deepcopy
 
-from src.strategy.asmodel import PureMM
-from src.broker.order_management_system import OrderManagementSystem
-from src.data.data_type import DataOrder, PriceSize, PositionSide, OrderType, Tickdata
-from src.data.inventory_management import InventoryManagement
-from src.data.history_management import HistoricalOrderDataManagement, HistoricalTickdata
+from deepmm.strategy.asmodel import PureMM
+from deepmm.broker.order_management_system import OrderManagementSystem
+from deepmm.data.data_type import DataOrder, PriceSize, PositionSide, OrderType, Tickdata
+from deepmm.data.inventory_management import InventoryManagement
+from deepmm.data.history_management import HistoricalOrderDataManagement, HistoricalTickdata
 
-from utils.date_management import check_stringtime_greater_closetime, \
+from deepmm.utils.date_management import check_stringtime_greater_closetime, \
                                     check_two_stringtime_greater_thresh, \
                                     check_stringtime_less_starttime, \
                                     check_two_string_is_same_day, \

src/broker/order_management_system.py → src/deepmm/broker/order_management_system.py

@@ -1,5 +1,5 @@
 from copy import deepcopy
-from src.data.data_type import DataOrder, PriceSize, PositionSide, OrderType
+from deepmm.data.data_type import DataOrder, PositionSide, OrderType
 
 class OrderManagementSystem:
     def __init__(self):

src/data/history_management.py → src/deepmm/data/history_management.py

@@ -1,8 +1,8 @@
 from copy import deepcopy
 import numpy as np
 import pandas as pd
-from src.data.data_type import PositionSide
-from utils.date_management import make_date_from_string
+from deepmm.data.data_type import PositionSide
+from deepmm.utils.date_management import make_date_from_string
 
 class HistoricalTickdata():
     def __init__(self):

src/data/inventory_management.py → src/deepmm/data/inventory_management.py

@@ -1,7 +1,6 @@
-import pandas as pd
-
 from copy import deepcopy
-from src.data.data_type import DataOrder, PositionSide, PriceSize
+from deepmm.data.data_type import DataOrder, PositionSide
+
 
 class InventoryManagement():
     def __init__(self, maximum_inventory):

src/optimizers/bruteforce_optimizer.py → src/deepmm/optimizers/bruteforce_optimizer.py

@@ -1,7 +1,7 @@
 import numpy as np
 from tqdm import tqdm
 import multiprocessing as mp
-from utils.path_management import increment_path
+
 
 def optimize_worker(datasets, params, run_dataset):
             profit, sharpe, mdd = run_dataset(datasets, 

src/pipeline.py → src/deepmm/pipeline.py

@@ -1,25 +1,26 @@
 import os
 import time
 import pytz
-import json
 import logging
-import optuna
 import numpy as np
 import pandas as pd
 from datetime import datetime
-from pathlib import Path
 
-from src.bot.bot import Bot
-from src.optimizers.bruteforce_optimizer import BruteForceOptimizer
-from src.data.data_type import Tickdata
-from src.metrics import sharpe_ratio, maximum_drawdown
-from src.utils.visualizer import VISUALIZER
+from deepmm.bot.bot import Bot
+from deepmm.optimizers.bruteforce_optimizer import BruteForceOptimizer
+from deepmm.data.data_type import Tickdata
+from deepmm.metrics import sharpe_ratio, maximum_drawdown
+from deepmm.utils.visualizer import VISUALIZER
 
 from utils.file_management import load_csv
 from utils.date_management import make_date_to_tickersymbol
 from utils.path_management import increment_path
 from utils.file_management import write_yaml
 
+from plutus.core.instrument import Instrument
+from plutus.datahub.redis_datahub import RedisDataHub, RedisDataHandler, InternalDataHubQuote
+
+
 TIMEZONE = pytz.timezone('Asia/Ho_Chi_Minh')
 class Pipeline():
     def __init__(self, opts):
@@ -29,6 +30,11 @@ def __init__(self, opts):
             self.val_data = load_csv(opts['DATASET']['VAL']['csv_file'])
             self.test_data = load_csv(opts['DATASET']['TEST']['csv_file'])
 
+        self.logger = None
+        self.current_symbol = None
+        self.model = None
+        self.visualizer = None
+
     def _init_logging(self, log_file='log.txt', name='logger'):
         """ Initialize logging
         """
@@ -90,15 +96,19 @@ def report_monthly_data(self, model, save_dir=None):
     def export_df_result(self, model, save_dir=None):
         totalAvgSpread, totalProfit, num_order = model.get_total_history().get_statistic()
 
-        profit, sharpe, max_drawdown_value = self.calculate_performance_score(totalProfit, 
-                                                                              num_order,
-                                                                              self.opts['PIPELINE']['params']['fee'])
+        profit, sharpe, max_drawdown_value = self.calculate_performance_score(
+            totalProfit,
+            num_order,
+            self.opts['PIPELINE']['params']['fee']
+        )
 
-        df = pd.DataFrame({'avg_spread': [totalAvgSpread],
-                           'num_trade': [num_order.sum()],
-                           'profit': [profit],
-                           'sharpe_ratio': [sharpe],
-                           'max_drawdown': [max_drawdown_value]})
+        df = pd.DataFrame({
+            'avg_spread': [totalAvgSpread],
+            'num_trade': [num_order.sum()],
+            'profit': [profit],
+            'sharpe_ratio': [sharpe],
+            'max_drawdown': [max_drawdown_value]
+        })
 
         if save_dir:
             df.to_csv(save_dir/'result.csv', index=False)
@@ -197,50 +207,53 @@ def run_dataset(self, datasets, type_data='train', is_visualize=True, params=Non
         logger.info(f"result on {type_data} with profit_return {profit}    sharpe {sharpe}   mdd {max_drawdown_value}")
         return profit, sharpe, max_drawdown_value
 
-
-    def run_papertrading(self, redis_client):
+    def data_handler_func(self, instrument: Instrument, internal_data_quote: InternalDataHubQuote):
+        cur_price = internal_data_quote.latest_matched_price
+
+        now = datetime.fromtimestamp(internal_data_quote.timestamp).astimezone(TIMEZONE)
+        if cur_price is None:
+            self.logger.info(f"There is no price update yet at {now.strftime('%Y-%m-%d %H:%M:%S')}")
+            return
+
+        instrument_str = str(instrument)
+
+        if self.current_symbol is None:
+            self.current_symbol = instrument_str
+            self.model.init_capacity_every_month()
+        elif self.current_symbol != instrument_str:
+            self.visualizer.visualize_monthly_data(
+                bot_data=self.model.get_monthly_history(),
+                bot_data_market_time_price=self.model.monthly_tick_data,
+                symbol=self.current_symbol,
+                save_dir=self.opts['PIPELINE']['params']['save_dir'] / 'papertrading'
+            )
+            self._log_results(self.logger, self.model, self.current_symbol)
+            self.report_monthly_data(
+                self.model,
+                save_dir=self.opts['PIPELINE']['params']['save_dir'] / 'papertrading' / self.current_symbol
+            )
+            self.model.init_capacity_every_month()
+
+        self.model.fit_tickdata(Tickdata(now, cur_price))
+
+    def run_papertrading(self, redis_datahub: RedisDataHub):
         os.makedirs(self.opts['PIPELINE']['params']['save_dir'], exist_ok=True)
-        logger = self._init_logging(self.opts['PIPELINE']['params']['save_dir']/f'papertrading_log.txt', name='papertrading_logger')
-        model = Bot(self.opts['PIPELINE']['params'], logger=logger)
-        visualizer = VISUALIZER(fees=self.opts['PIPELINE']['params']['fee'])
+        self.logger = self._init_logging(self.opts['PIPELINE']['params']['save_dir']/f'papertrading_log.txt', name='papertrading_logger')
+        self.model = Bot(self.opts['PIPELINE']['params'], logger=self.logger)
+        self.visualizer = VISUALIZER(fees=self.opts['PIPELINE']['params']['fee'])
         current_date = datetime.now()
-        tickersymbol = make_date_to_tickersymbol(current_date)
-        current_symbol = None
-        logger.info(f"Start papertrading with tickersymbol {tickersymbol}")
-        logger.info(f"with parameters: {self.opts['PIPELINE']['params']}")
-        def redis_message_handler(redis_message, current_symbol=current_symbol, model=model, logger=logger, visualizer=visualizer, tickersymbol=tickersymbol):
-
-            quote = json.loads(redis_message['data'])
-            cur_price = quote['latest_matched_price']
-
-            now = datetime.fromtimestamp(quote['timestamp']).astimezone(TIMEZONE)
-            if cur_price is None:
-                logger.info(f"There is no price update yet at {now.strftime('%Y-%m-%d %H:%M:%S')}")
-                return
-
-            if current_symbol is None:
-                current_symbol = tickersymbol
-                model.init_capacity_every_month()   
-            elif current_symbol != tickersymbol:
-                visualizer.visualize_monthly_data(bot_data=model.get_monthly_history(),
-                                                  bot_data_market_time_price=model.monthly_tick_data,
-                                                  symbol=current_symbol,
-                                                  save_dir=self.opts['PIPELINE']['params']['save_dir']/'papertrading')
-                self._log_results(logger,model, current_symbol)
-                self.report_monthly_data(model, save_dir=self.opts['PIPELINE']['params']['save_dir']/'papertrading'/current_symbol)
-                model.init_capacity_every_month()
-
-            model.fit_tickdata(Tickdata(now, cur_price))
-
-        F1M_CHANNEL = f'HNXDS:{tickersymbol}'
-
-        pub_sub = redis_client.pubsub()
-        pub_sub.psubscribe(**{F1M_CHANNEL: redis_message_handler})
-        # subcribe to channel F1M channel
-        # register a callback function to handle message received from redis-server
-        while True:
-            pubsub_thread = pub_sub.run_in_thread(sleep_time=1)
-
-            time.sleep(60)
-
-            # pubsub_thread.stop()
+        ticker_symbol = make_date_to_tickersymbol(current_date)
+        instrument = Instrument(ticker_symbol=ticker_symbol, exchange_code_str='HNXDS')
+        self.logger.info(f"Start papertrading with tickersymbol {ticker_symbol}")
+        self.logger.info(f"with parameters: {self.opts['PIPELINE']['params']}")
+
+        f1m_channel_pattern = instrument.id
+
+        data_handler = RedisDataHandler(
+            data_handler_function=self.data_handler_func,
+            subscribed_pattern=f1m_channel_pattern,
+            run_in_thread=True,
+            sleep_time=0.001,
+        )
+        redis_datahub.data_handler_list.append(data_handler)
+        redis_datahub.start_pubsub()

run.py → src/deepmm/run.py

@@ -1,6 +1,6 @@
 from pathlib import Path
 
-from src.pipeline import Pipeline
+from deepmm.pipeline import Pipeline
 from utils.argument_management import Opts
 
 FILE = Path(__file__).resolve()

run_papertrading.py → src/deepmm/run_papertrading.py

@@ -1,37 +1,34 @@
-import redis
+import time
+
 import json
 import pytz
 from datetime import datetime
 from pathlib import Path
 
-
-from src.pipeline import Pipeline
+from pipeline import Pipeline
 from utils.date_management import make_date_to_tickersymbol
 from utils.argument_management import Opts
 from utils.file_management import load_yaml
 
+from plutus.datahub.redis_datahub import RedisDataHub
 
 FILE = Path(__file__).resolve()
 ROOT = FILE.parents[0]
 
 TIMEZONE = pytz.timezone('Asia/Ho_Chi_Minh')
 
+
 # handle pricehub quote
 def redis_message_handler(redis_message):
 
-  quote = json.loads(redis_message['data'])
-  cur_price = quote['latest_matched_price']
-
-  # check if cur_price updated yet
-  if cur_price is None:
-    return
-
-  datetime_now = datetime.fromtimestamp(quote['timestamp']).astimezone(TIMEZONE).time()
+    quote = json.loads(redis_message['data'])
+    cur_price = quote['latest_matched_price']
 
+    # check if cur_price updated yet
+    if cur_price is None:
+        return
 
-
-
-
+    datetime_now = datetime.fromtimestamp(quote['timestamp']).astimezone(TIMEZONE).time()
 
 
 def main():
@@ -41,24 +38,22 @@ def main():
     redis_opts = load_yaml(ROOT/'configs'/'usr'/'redis_account.yaml')
     redis_host, redis_port, redis_password = redis_opts['host'], redis_opts['port'], redis_opts['password']
     print(redis_host, redis_port, redis_password)
-
-    # connect to redis server
-    redis_client = redis.Redis(
-    host=redis_host,
-    port=redis_port,
-    password=redis_password
-    )
 
-    # check connection to redis OK
-    print(redis_client.ping())
+    redis_data_hub = RedisDataHub(
+        redis_host=redis_host,
+        redis_port=redis_port,
+        redis_password=redis_password
+    )
 
     current_date = datetime.now()
-    tickersymbol = make_date_to_tickersymbol(current_date)
-    F1M_CHANNEL = f'HNXDS:{tickersymbol}'
+    ticker_symbol = make_date_to_tickersymbol(current_date)
+    F1M_CHANNEL = f'HNXDS:{ticker_symbol}'
     print(F1M_CHANNEL)
 
-    pipeline.run_papertrading(redis_client)
+    pipeline.run_papertrading(redis_data_hub)
+
 
 if __name__ == "__main__":
     main()
-
+    while True:
+        time.sleep(0.01)

src/strategy/asmodel.py → src/deepmm/strategy/asmodel.py

@@ -2,7 +2,6 @@
 import numpy as np
 import math
 
-from utils.date_management import calculate_distance_milis
 
 class PureMM:
     def __init__(self, opts):

utils/argument_management.py → src/deepmm/utils/argument_management.py

@@ -6,8 +6,8 @@
 import yaml
 import json
 from pathlib import Path
-from utils.file_management import load_yaml
-from utils.path_management import increment_path
+from deepmm.utils.file_management import load_yaml
+from deepmm.utils.path_management import increment_path
 
 
 class Config(dict):

utils/download.py → src/deepmm/utils/download.py

@@ -1,4 +1,4 @@
-from utils.db_connection import DataConnection
+from deepmm.utils.db_connection import DataConnection
 import os
 
 

download_data_from_db.py → src/deepmm/utils/download_data_from_db.py

@@ -1,12 +1,8 @@
 from pathlib import Path
 
-from utils.date_management import get_num_days_to_maturity
+from deepmm.utils.db_connection import DataConnection
 
-from utils.db_connection import DataConnection
-
-from utils.file_management import load_yaml
-from utils.download import down_derivative_midprice_db, down_derivative_matched_db
-import pandas as pd
+from deepmm.utils.file_management import load_yaml
 
 FILE = Path(__file__).resolve()
 ROOT = FILE.parents[0]

src/utils/visualizer.py → src/deepmm/utils/visualizer.py

@@ -7,7 +7,7 @@
 
 import plotly.subplots
 
-from src.data.history_management import HistoricalOrderDataManagement, HistoricalTickdata
+from deepmm.data.history_management import HistoricalOrderDataManagement, HistoricalTickdata
 
 
 class VISUALIZER():