Timestamp Unittests

precog/miners/base_miner.py

@@ -6,7 +6,7 @@
 
 from precog.protocol import Challenge
 from precog.utils.cm_data import CMData
-from precog.utils.timestamp import datetime_to_CM_timestamp, iso8601_to_datetime
+from precog.utils.timestamp import datetime_to_iso8601, iso8601_to_datetime
 
 
 def get_point_estimate(timestamp: str) -> float:
@@ -23,8 +23,8 @@ def get_point_estimate(timestamp: str) -> float:
 
     # Set the time range to be as small as possible for query speed
     # Set the start time as 2 seconds prior to the provided time
-    start_time: str = datetime_to_CM_timestamp(iso8601_to_datetime(timestamp) - timedelta(days=1))
-    end_time: str = datetime_to_CM_timestamp(iso8601_to_datetime(timestamp))  # built-ins handle CM API's formatting
+    start_time: str = datetime_to_iso8601(iso8601_to_datetime(timestamp) - timedelta(seconds=2))
+    end_time: str = datetime_to_iso8601(iso8601_to_datetime(timestamp))
 
     # Query CM API for a pandas dataframe with only one record
     price_data: pd.DataFrame = cm.get_CM_ReferenceRate(assets="BTC", start=start_time, end=end_time)
@@ -56,8 +56,8 @@ def get_prediction_interval(timestamp: str, point_estimate: float) -> Tuple[floa
     cm = CMData()
 
     # Set the time range to be 24 hours
-    start_time: str = datetime_to_CM_timestamp(iso8601_to_datetime(timestamp) - timedelta(days=1))
-    end_time: str = datetime_to_CM_timestamp(iso8601_to_datetime(timestamp))  # built-ins handle CM API's formatting
+    start_time: str = datetime_to_iso8601(iso8601_to_datetime(timestamp) - timedelta(days=1))
+    end_time: str = datetime_to_iso8601(iso8601_to_datetime(timestamp))  # built-ins handle CM API's formatting
 
     # Query CM API for sample standard deviation of the 1s residuals
     historical_price_data: pd.DataFrame = cm.get_CM_ReferenceRate(

precog/utils/timestamp.py

@@ -66,14 +66,14 @@ def datetime_to_iso8601(timestamp: datetime) -> str:
     """
     Convert datetime to iso 8601 string
     """
-    return timestamp.isoformat()
+    return timestamp.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
 
 
 def iso8601_to_datetime(timestamp: str) -> datetime:
     """
     Convert iso 8601 string to datetime
     """
-    return datetime.fromisoformat(timestamp)
+    return datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=get_timezone())
 
 
 def posix_to_datetime(timestamp: float) -> datetime:
@@ -83,13 +83,6 @@ def posix_to_datetime(timestamp: float) -> datetime:
     return datetime.fromtimestamp(timestamp, tz=get_timezone())
 
 
-def datetime_to_CM_timestamp(timestamp: datetime) -> str:
-    """
-    Convert iso 8601 string to coinmetrics timestamp
-    """
-    return timestamp.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
-
-
 ###############################
 #          FUNCTIONS          #
 ###############################

precog/validators/reward.py

@@ -8,7 +8,7 @@
 from precog.protocol import Challenge
 from precog.utils.cm_data import CMData
 from precog.utils.general import pd_to_dict, rank
-from precog.utils.timestamp import align_timepoints, datetime_to_CM_timestamp, iso8601_to_datetime, mature_dictionary
+from precog.utils.timestamp import align_timepoints, datetime_to_iso8601, iso8601_to_datetime, mature_dictionary
 
 
 ################################################################################
@@ -24,8 +24,8 @@ def calc_rewards(
     decayed_weights = decay**weights
     timestamp = responses[0].timestamp
     cm = CMData()
-    start_time: str = datetime_to_CM_timestamp(iso8601_to_datetime(timestamp) - timedelta(hours=1))
-    end_time: str = datetime_to_CM_timestamp(iso8601_to_datetime(timestamp))  # built-ins handle CM API's formatting
+    start_time: str = datetime_to_iso8601(iso8601_to_datetime(timestamp) - timedelta(hours=1))
+    end_time: str = datetime_to_iso8601(iso8601_to_datetime(timestamp))  # built-ins handle CM API's formatting
     # Query CM API for sample standard deviation of the 1s residuals
     historical_price_data: DataFrame = cm.get_CM_ReferenceRate(
         assets="BTC", start=start_time, end=end_time, frequency="1s"

pyproject.toml

@@ -38,6 +38,7 @@ isort = "^5.13.2"
 mypy = "^1.13.0"
 flake8 = "^7.1.1"
 pytest = "^8.3.3"
+hypothesis = "^6.122.3"
 
 
 [tool.black]

tests/test_package.py

@@ -5,6 +5,11 @@
 
 class TestPackage(unittest.TestCase):
 
+    # runs once prior to all tests in this file
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+
+    # runs once prior to every single test
     def setUp(self):
         pass
 

tests/utils/test_timestamp.py

@@ -0,0 +1,161 @@
+import unittest
+from datetime import datetime
+
+from hypothesis import given
+from hypothesis import strategies as st
+from pytz import timezone
+
+from precog.utils.timestamp import (
+    datetime_to_iso8601,
+    datetime_to_posix,
+    get_midnight,
+    get_now,
+    get_posix,
+    get_timezone,
+    iso8601_to_datetime,
+    posix_to_datetime,
+)
+
+
+class TestTimestamp(unittest.TestCase):
+
+    # runs once prior to all tests in this file
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+
+        self.DATETIME_CONSTANT = datetime(2024, 12, 11, 18, 46, 43, 112378, tzinfo=get_timezone())
+        self.POSIX_CONSTANT = 1733942803.112378
+        self.ISO8601_CONSTANT = "2024-12-11T18:46:43.112378Z"
+
+    # runs once prior to every single test
+    def setUp(self):
+        pass
+
+    def test_get_timezone(self):
+        # Ensure we are abiding by UTC timezone
+        self.assertEqual(get_timezone(), timezone("UTC"))
+
+    def test_get_now(self):
+        now = get_now()
+
+        # Check that this is a datetime
+        self.assertIsInstance(now, datetime)
+
+        # Check that this is UTC timezone aware
+        self.assertEqual(now.tzinfo, get_timezone())
+
+        # Check that the timestamp is sensitive
+        now2 = get_now()
+        self.assertNotEqual(now, now2)
+
+        # Make our own timestamp
+        real_now = datetime.now(get_timezone())
+
+        # Check that all timezones are within 20 seconds of each other
+        # They should all be very close together considering everything
+        threshold = 20
+        diff1 = (now - now2).total_seconds()
+        diff2 = (now - real_now).total_seconds()
+        diff3 = (now2 - real_now).total_seconds()
+
+        self.assertLess(abs(diff1), threshold)
+        self.assertLess(abs(diff2), threshold)
+        self.assertLess(abs(diff3), threshold)
+
+    def test_get_posix(self):
+        posix = get_posix()
+
+        # Check that this is a float
+        self.assertIsInstance(posix, float)
+
+        # Check that the timestamp is sensitive
+        posix2 = get_posix()
+        self.assertNotEqual(posix, posix2)
+
+        # Make our own timestamp
+        real_posix = datetime.now(get_timezone()).timestamp()
+
+        # Check that all timezones are within 20 seconds of each other
+        # They should all be very close together considering everything
+        threshold = 20
+        diff1 = posix - posix2
+        diff2 = posix - real_posix
+        diff3 = posix2 - real_posix
+
+        self.assertLess(abs(diff1), threshold)
+        self.assertLess(abs(diff2), threshold)
+        self.assertLess(abs(diff3), threshold)
+
+    def test_get_midnight(self):
+        now = datetime.now(get_timezone())
+        midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
+
+        midnight2 = get_midnight()
+
+        # Check that this is UTC timezone aware
+        self.assertEqual(midnight2.tzinfo, get_timezone())
+
+        # They should be equal
+        self.assertEqual(midnight, midnight2)
+
+        # confirm everything is zeroed out
+        self.assertEqual(midnight2.microsecond, 0)
+        self.assertEqual(midnight2.second, 0)
+        self.assertEqual(midnight2.minute, 0)
+        self.assertEqual(midnight2.hour, 0)
+
+    def test_datetime_iso_roundtrip(self):
+        # datetime -> str -> datetime
+        new_str = datetime_to_iso8601(self.DATETIME_CONSTANT)
+        new_datetime = iso8601_to_datetime(new_str)
+
+        self.assertEqual(self.DATETIME_CONSTANT, new_datetime)
+        self.assertEqual(self.ISO8601_CONSTANT, new_str)
+        self.assertEqual(new_datetime.tzinfo, get_timezone())
+
+        # str -> datetime -> str
+        new_datetime = iso8601_to_datetime(self.ISO8601_CONSTANT)
+        new_str = datetime_to_iso8601(new_datetime)
+
+        self.assertEqual(self.DATETIME_CONSTANT, new_datetime)
+        self.assertEqual(self.ISO8601_CONSTANT, new_str)
+        self.assertEqual(new_datetime.tzinfo, get_timezone())
+
+    @given(st.datetimes(timezones=st.just(get_timezone())))
+    def test_hypothesis_datetime_iso_roundtrip(self, new_datetime):
+        # datetime -> str -> datetime
+        new_str = datetime_to_iso8601(new_datetime)
+        new_datetime2 = iso8601_to_datetime(new_str)
+
+        self.assertEqual(new_datetime, new_datetime2)
+        self.assertEqual(new_datetime2.tzinfo, get_timezone())
+
+    def test_datetime_posix_roundtrip(self):
+        # datetime -> float -> datetime
+        new_float = datetime_to_posix(self.DATETIME_CONSTANT)
+        new_datetime = posix_to_datetime(new_float)
+
+        self.assertEqual(self.DATETIME_CONSTANT, new_datetime)
+        self.assertEqual(self.POSIX_CONSTANT, new_float)
+        self.assertEqual(new_datetime.tzinfo, get_timezone())
+
+        # float -> datetime -> float
+        new_datetime = posix_to_datetime(self.POSIX_CONSTANT)
+        new_float = datetime_to_posix(new_datetime)
+
+        self.assertEqual(self.DATETIME_CONSTANT, new_datetime)
+        self.assertEqual(self.POSIX_CONSTANT, new_float)
+        self.assertEqual(new_datetime.tzinfo, get_timezone())
+
+    @given(st.datetimes(timezones=st.just(get_timezone())).map(lambda dt: dt.replace(microsecond=0)))
+    def test_hypothesis_datetime_posix_roundtrip(self, new_datetime):
+        # hypothesis exposes niche floating point precision errors
+        # zero out microseconds to solve this
+        # Floating point precision resulting in a mismatch of 1 microsecond is negligible
+
+        # datetime -> float -> datetime
+        new_float = float(datetime_to_posix(new_datetime))
+        new_datetime2 = posix_to_datetime(new_float)
+
+        self.assertEqual(new_datetime, new_datetime2)
+        self.assertEqual(new_datetime2.tzinfo, get_timezone())