feat: add 'Signal' message function and tests

src/rpc/utils/messages.py

@@ -57,7 +57,7 @@ def get_ohlcv_message(
     volume: float,
 ) -> candlestick_struct_pb2.Ohlcv:
     """
-        Returns a Ohlcv message
+        Returns a Ohlcv message.
     
         Parameters:
             - open (float): Opening price
@@ -76,3 +76,36 @@ def get_ohlcv_message(
         close = close,
         volume = volume,
     )
+
+
+def get_signal_message(
+    timestamp: datetime,
+    position: str,
+    side: str,
+    take_profit: float,
+    stop_loss: float,
+    confidence: float
+) -> candlestick_struct_pb2.Signal:
+    """
+        Returns a Signal message.
+
+        Parameters:
+            - timestamp (datetime): Timestamp of the signal.
+            - position (str): Position of the signal.
+            - side (str): Side of the signal.
+            - take_profit (float): Take profit value.
+            - stop_loss (float): Stop loss value.
+            - confidence (float): Confidence level of the signal.
+
+        Returns:
+            - (candlestick_struct_pb2.Signal): Signal message.
+    """
+    timestamp = get_timestamp_message(timestamp)    # Python datetime to `google.protobuf.Timestamp`
+    return candlestick_struct_pb2.Signal(
+        timestamp = timestamp,
+        position = position,
+        side = side,
+        take_profit = take_profit,
+        stop_loss = stop_loss,
+        confidence = confidence,
+    )

tests/__init__.py

@@ -0,0 +1,2 @@
+import sys
+sys.path.append('protos')

tests/test_candlestick.py

@@ -9,12 +9,14 @@
 
 import unittest
 from datetime import datetime
+from protos import candlestick_struct_pb2
 
 # Third-party
 from src.rpc.utils.messages import (
     get_ohlcv_message,
     get_timestamp_message,
     get_candlestick_message,
+    get_signal_message,
 )
 
 
@@ -72,6 +74,35 @@ def test_get_candlestick_message(self):
         assert candlestick.timestamp == timestamp, 'invalid `timestamp` value'
 
 
+    def test_get_signal_message(self):
+        """ Test for `get_signal_message` """
+        _datetime = datetime(
+            year = 2024,
+            month = 3,
+            day = 20,
+            hour = 12,
+            minute = 20,
+            second = 45,
+            microsecond = 150
+        )
+        take_profit = 110
+        stop_loss = 90
+        confidence = 0.9
+        signal = get_signal_message(
+            timestamp = _datetime,
+            position = 'LONG',
+            side = 'BUY',
+            take_profit = take_profit,
+            stop_loss = stop_loss,
+            confidence = confidence
+        )
+        # Test
+        assert signal.timestamp.seconds == 1710937245, 'invalid seconds value'
+        assert signal.timestamp.nanos == 150000, 'invalid nanoseconds value'
+        assert signal.take_profit == take_profit, 'invalid take_profit value'
+        assert signal.stop_loss == stop_loss, 'invalid stop_loss value'
+        assert signal.confidence == confidence, 'invalid confidence value'
+
 
 if __name__ == '__main__':
     unittest.main(exit=False)