diff --git a/src/rpc/utils/messages.py b/src/rpc/utils/messages.py index 753a2bf..2b75dd6 100644 --- a/src/rpc/utils/messages.py +++ b/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, + ) diff --git a/tests/__init__.py b/tests/__init__.py index e69de29..c7cf38e 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -0,0 +1,2 @@ +import sys +sys.path.append('protos') \ No newline at end of file diff --git a/tests/test_candlestick.py b/tests/test_candlestick.py index 22db39b..1009633 100644 --- a/tests/test_candlestick.py +++ b/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)