mlflow service intgrated with the backend

app/services/backtest_service.py

@@ -7,7 +7,8 @@ def run_backtest_by_id(backtest_id):
     backtest = Backtest.query.get(backtest_id)
     if not backtest:
         return
-
+
+
     # Simulate backtest processing
     result = Result(
         backtest_id=backtest_id,

app/services/kafka_service.py

@@ -2,6 +2,7 @@
 from confluent_kafka.admin import AdminClient, NewTopic
 from confluent_kafka import Producer, Consumer, KafkaException, KafkaError
 import json
+from decimal import Decimal
 
 class KafkaService:
     def __init__(self, brokers):
@@ -29,9 +30,15 @@ def create_topic(self, topic):
         else:
             logging.info(f"Topic {topic} already exists")
 
+    def json_serializer(self, obj):
+        if isinstance(obj, Decimal):
+            return float(obj)
+        raise TypeError("Type not serializable")
+
     def produce(self, topic, message):
         logging.info(f"Producing message to topic {topic}: {message}")
-        self.producer.produce(topic, key=None, value=json.dumps(message))
+        serialized_message = json.dumps(message, default=self.json_serializer)
+        self.producer.produce(topic, key=None, value=serialized_message)
         self.producer.flush()
         logging.info("Message produced successfully")
 

app/services/mlflow_service.py

@@ -10,4 +10,4 @@ def log_metrics(self, run_name, metrics):
             for key, value in metrics.items():
                 mlflow.log_metric(key, value)
 
-mlflow_service = MLflowService(tracking_uri='http://localhost:5000')
+mlflow_service = MLflowService(tracking_uri='http://localhost:5050')