feat: add tests and enums

.gitignore

@@ -160,5 +160,4 @@ cython_debug/
 #.idea/
 
 # other
-protos/**/*_pb2.py
-protos/**/*_pb2_grpc.py
+protos/**/*.py

protos/enums_struct.proto

@@ -0,0 +1,23 @@
+syntax = "proto3";
+
+// [enum] Position
+// Represents a position, either long, short.
+enum Position {
+    // No position.
+    UNKNOWN_POSITION = 0;
+    // Long position.
+    LONG = 1;
+    // Short position.
+    SHORT = 2;
+}
+
+// [enum] Side
+// Represents a side, either buy, sell.
+enum Side {
+    // No position.
+    UNKNOWN_SIDE = 0;
+    // Buy side.
+    BUY = 1;
+    // Sell side.
+    SELL = 2;
+}

scripts/test/run.sh

@@ -0,0 +1,2 @@
+#!/bin/bash
+python3 -u -m unittest

src/rpc/utils/messages.py

@@ -0,0 +1,78 @@
+##########################################################
+#
+# Copyright (C) 2023-PRESENT: Keivan Ipchi Hagh
+#
+# Email:    [email protected]
+# GitHub:   https://github.com/keivanipchihagh
+#
+##########################################################
+
+from datetime import datetime
+from google.protobuf import timestamp_pb2
+from protos import candlestick_struct_pb2
+
+
+def get_timestamp_message(
+    timestamp: datetime,
+) -> timestamp_pb2.Timestamp:
+    """
+        Returns a `google.protobuf.Timestamp` message from Python datetime object.
+
+        Parameters:
+            - timestamp (datetime): Datetime object representing the timestamp.
+
+        Returns:
+            - (timestamp_pb2.Timestamp): Timestamp message.
+    """
+    timestamp_message = timestamp_pb2.Timestamp()
+    timestamp_message.FromDatetime(timestamp)
+    return timestamp_message
+
+
+def get_candlestick_message(
+    timestamp: timestamp_pb2.Timestamp,
+    ohlcv: candlestick_struct_pb2.Ohlcv
+) -> candlestick_struct_pb2.Candlestick:
+    """
+        Returns a Candlestick message.
+
+        Parameters:
+            - timestamp (timestamp_pb2.Timestamp): Timestamp of the candlestick.
+            - ohlcv (candlestick_struct_pb2.Ohlcv): OHLCV message.
+
+        Returns:
+            - (candlestick_struct_pb2.Candlestick): Candlestick message.
+    """
+    return candlestick_struct_pb2.Candlestick(
+        timestamp = timestamp,
+        ohlcv = ohlcv,
+    )
+
+
+def get_ohlcv_message(
+    open: float,
+    high: float,
+    low: float,
+    close: float,
+    volume: float,
+) -> candlestick_struct_pb2.Ohlcv:
+    """
+        Returns a Ohlcv message
+    
+        Parameters:
+            - open (float): Opening price
+            - high (float): Highest price.
+            - low (float): Lowest price.
+            - close (float): Closing price.
+            - volume (float): Candlestick volume.
+
+        Returns:
+            - (candlestick_struct_pb2.Ohlcv): OHLCV message
+    """
+    return candlestick_struct_pb2.Ohlcv(
+        open = open,
+        high = high,
+        low =  low,
+        close = close,
+        volume = volume,
+    )

tests/test_candlestick.py

@@ -0,0 +1,77 @@
+##########################################################
+#
+# Copyright (C) 2023-PRESENT: Keivan Ipchi Hagh
+#
+# Email:    [email protected]
+# GitHub:   https://github.com/keivanipchihagh
+#
+##########################################################
+
+import unittest
+from datetime import datetime
+
+# Third-party
+from src.rpc.utils.messages import (
+    get_ohlcv_message,
+    get_timestamp_message,
+    get_candlestick_message,
+)
+
+
+class TestCandlestick(unittest.TestCase):
+    """ Tests for `Candlestick` """
+
+
+    def test_get_timestamp_message(self):
+        """ Test for `get_timestamp_message` """
+        _datetime = datetime(
+            year = 2024,
+            month = 3,
+            day = 20,
+            hour = 12,
+            minute = 20,
+            second = 45,
+            microsecond = 150
+        )
+        timestamp = get_timestamp_message(_datetime)
+        # Test
+        assert timestamp.seconds == 1710937245, 'invalid seconds value'
+        assert timestamp.nanos == 150000, 'invalid nanoseconds value'
+        # Return value
+        return timestamp
+
+
+    def test_get_ohlcv_message(self):
+        """ Test for `get_ohlcv_message` """
+        open = 100
+        high = 110
+        low = 90
+        close = 80
+        volume = 1000
+        ohlcv = get_ohlcv_message(open, high, low, close, volume)
+        # Test
+        assert ohlcv.open == open, 'invalid `open` field value'
+        assert ohlcv.high == high, 'invalid `high` field value'
+        assert ohlcv.low == low, 'invalid `low` field value'
+        assert ohlcv.close == close, 'invalid `close` field value'
+        assert ohlcv.volume == volume, 'invalid `volume` field value'
+        # Return value
+        return ohlcv
+
+
+    def test_get_candlestick_message(self):
+        """ Test for `get_candlestick_message` """
+        timestamp = self.test_get_timestamp_message()
+        ohlcv = self.test_get_ohlcv_message()
+        candlestick = get_candlestick_message(
+            timestamp = timestamp,
+            ohlcv = ohlcv
+        )
+        # Test
+        assert candlestick.ohlcv == ohlcv, 'invalid `ohlcv` value'
+        assert candlestick.timestamp == timestamp, 'invalid `timestamp` value'
+
+
+
+if __name__ == '__main__':
+    unittest.main(exit=False)