Timestamp Unittests

precog/miners/base_miner.py

@@ -1,12 +1,11 @@
-from datetime import timedelta
 from typing import Tuple
 
 import bittensor as bt
 import pandas as pd
 
 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 get_before, to_datetime, to_str
 
 
 def get_point_estimate(timestamp: str) -> float:
@@ -21,13 +20,13 @@ def get_point_estimate(timestamp: str) -> float:
     # Create data gathering instance
     cm = CMData()
 
-    # 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
+    # Ensure timestamp is correctly typed and set to UTC
+    provided_timestamp = 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)
+    price_data: pd.DataFrame = cm.get_CM_ReferenceRate(
+        assets="BTC", start=None, end=to_str(provided_timestamp), frequency="1s", limit_per_asset=1, paging_from="end"
+    )
 
     # Get current price closest to the provided timestamp
     btc_price: float = float(price_data["ReferenceRateUSD"].iloc[-1])
@@ -56,22 +55,23 @@ 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
+    # Ensure both timestamps are correctly typed and set to UTC
+    start_time = get_before(timestamp, days=1, minutes=0, seconds=0)
+    end_time = to_datetime(timestamp)
 
     # Query CM API for sample standard deviation of the 1s residuals
     historical_price_data: pd.DataFrame = cm.get_CM_ReferenceRate(
-        assets="BTC", start=start_time, end=end_time, frequency="1s"
+        assets="BTC", start=to_str(start_time), end=to_str(end_time), frequency="1s"
     )
     residuals: pd.Series = historical_price_data["ReferenceRateUSD"].diff()
     sample_std_dev: float = float(residuals.std())
 
     # We have the standard deviation of the 1s residuals
-    # We are forecasting forward 5m, which is 300s
-    # We must scale the 1s sample standard deviation to reflect a 300s forecast
+    # We are forecasting forward 60m, which is 3600s
+    # We must scale the 1s sample standard deviation to reflect a 3600s forecast
     # Make reasonable assumptions that the 1s residuals are uncorrelated and normally distributed
     # To do this naively, we multiply the std dev by the square root of the number of time steps
-    time_steps: int = 300
+    time_steps: int = 3600
     naive_forecast_std_dev: float = sample_std_dev * (time_steps**0.5)
 
     # For a 90% prediction interval, we use the coefficient 1.64

precog/utils/classes.py

@@ -2,7 +2,7 @@
 from datetime import datetime, timedelta
 from typing import List
 
-from precog.utils.timestamp import get_now, get_timezone, iso8601_to_datetime, round_minute_down
+from precog.utils.timestamp import get_now, get_timezone, round_minute_down, to_datetime
 
 
 class Config:
@@ -66,7 +66,7 @@ def __init__(self, uid: int, timezone=get_timezone()):
 
     def add_prediction(self, timestamp, prediction: float, interval: List[float]):
         if isinstance(timestamp, str):
-            timestamp = iso8601_to_datetime(timestamp)
+            timestamp = to_datetime(timestamp)
         timestamp = round_minute_down(timestamp)
         if prediction is not None:
             self.predictions[timestamp] = prediction
@@ -86,7 +86,7 @@ def format_predictions(self, reference_timestamp=None, hours: int = 1):
         if reference_timestamp is None:
             reference_timestamp = round_minute_down(get_now())
         if isinstance(reference_timestamp, str):
-            reference_timestamp = iso8601_to_datetime(reference_timestamp)
+            reference_timestamp = to_datetime(reference_timestamp)
         start_time = round_minute_down(reference_timestamp) - timedelta(hours=hours + 1)
         filtered_pred_dict = {
             key: value for key, value in self.predictions.items() if start_time <= key <= reference_timestamp

precog/utils/timestamp.py

@@ -1,4 +1,5 @@
 from datetime import datetime, timedelta
+from typing import Optional, Union
 
 from pytz import timezone
 
@@ -21,12 +22,25 @@ def get_now() -> datetime:
     return datetime.now(get_timezone())
 
 
-def get_before(minutes: int = 5, seconds: int = 0) -> datetime:
+def get_before(
+    timestamp: Optional[Union[datetime, str, float]] = None,
+    days: int = 0,
+    hours: int = 0,
+    minutes: int = 5,
+    seconds: int = 0,
+) -> datetime:
     """
     Get the datetime x minutes before now
     """
-    now = get_now()
-    return now - timedelta(minutes=minutes, seconds=seconds)
+    if timestamp is None:
+        timestamp = get_now()
+    else:
+        timestamp = to_datetime(timestamp)
+
+    # Perform the time subtraction
+    before_timestamp = timestamp - timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
+
+    return before_timestamp
 
 
 def get_midnight() -> datetime:
@@ -40,54 +54,73 @@ def get_posix() -> float:
     """
     Get the current POSIX time, seconds that have elapsed since Jan 1 1970
     """
-    return datetime_to_posix(get_now())
+    return to_posix(get_now())
 
 
-def get_iso8601() -> str:
+def get_str() -> str:
     """
     Get the current timestamp as a string, convenient for requests
     """
-    return datetime_to_iso8601(get_now())
+    return to_str(get_now())
 
 
 ###############################
 #         CONVERTERS          #
 ###############################
 
 
-def datetime_to_posix(timestamp: datetime) -> float:
+def to_posix(timestamp: Union[datetime, str, float]) -> float:
     """
     Convert datetime to seconds that have elapsed since Jan 1 1970
     """
-    return timestamp.timestamp()
 
+    # Verify datetime object and convert to UTC
+    utc_datetime = to_datetime(timestamp)
+
+    # Convert to posix time float
+    posix_timestamp = utc_datetime.timestamp()
 
-def datetime_to_iso8601(timestamp: datetime) -> str:
+    return float(posix_timestamp)
+
+
+def to_str(timestamp: Union[datetime, str, float]) -> str:
     """
     Convert datetime to iso 8601 string
     """
-    return timestamp.isoformat()
 
+    # Verify datetime object and convert to UTC
+    utc_datetime = to_datetime(timestamp)
 
-def iso8601_to_datetime(timestamp: str) -> datetime:
-    """
-    Convert iso 8601 string to datetime
-    """
-    return datetime.fromisoformat(timestamp)
+    # Convert to iso8601 string
+    str_datetime = utc_datetime.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+
+    return str(str_datetime)
 
 
-def posix_to_datetime(timestamp: float) -> datetime:
+def to_datetime(timestamp: Union[str, float]) -> datetime:
     """
-    Convert seconds since Jan 1 1970 to datetime
+    Convert iso 8601 string, or a POSIX time float, to datetime
     """
-    return datetime.fromtimestamp(timestamp, tz=get_timezone())
+    if isinstance(timestamp, str):
+        # Assume the proper iso 8601 string format is used
+        # `strptime` will trigger an error as needed
+        return datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=get_timezone())
 
+    elif isinstance(timestamp, float):
+        # Assume proper float value
+        # `fromtimestamp` will trigger errors as needed
+        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")
+    elif isinstance(timestamp, datetime):
+        # Already a datetime object
+        # Return as UTC
+        return timestamp.astimezone(get_timezone())
+
+    else:
+        # Invalid typing
+        raise TypeError(
+            "Must pass a timestamp that is either a iso 8601 string, POSIX time float, or a datetime object"
+        )
 
 
 ###############################
@@ -128,8 +161,8 @@ def is_query_time(prediction_interval: int, timestamp: str, tolerance: int = 120
         First, check that we are in a new epoch
         Then, check if we already sent a request in the current epoch
     """
-    now = get_now()
-    provided_timestamp = iso8601_to_datetime(timestamp)
+    now: datetime = get_now()
+    provided_timestamp: datetime = to_datetime(timestamp)
 
     # The provided timestamp is the last time a request was made. If this timestamp
     # is from the current epoch, we do not want to make a request. One way to check

precog/validators/reward.py

@@ -1,4 +1,3 @@
-from datetime import timedelta  # temporary for dummy data
 from typing import List
 
 import bittensor as bt
@@ -8,7 +7,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, get_before, mature_dictionary, to_datetime, to_str
 
 
 ################################################################################
@@ -24,8 +23,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 = to_str(get_before(timestamp=timestamp, hours=1))
+    end_time: str = to_str(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"

precog/validators/weight_setter.py

@@ -12,14 +12,7 @@
 from precog.utils.bittensor import check_uid_availability, print_info, setup_bittensor_objects
 from precog.utils.classes import MinerHistory
 from precog.utils.general import func_with_retry, loop_handler
-from precog.utils.timestamp import (
-    datetime_to_iso8601,
-    elapsed_seconds,
-    get_before,
-    get_now,
-    is_query_time,
-    iso8601_to_datetime,
-)
+from precog.utils.timestamp import elapsed_seconds, get_before, get_now, get_str, is_query_time, to_datetime, to_str
 from precog.utils.wandb import log_wandb, setup_wandb
 from precog.validators.reward import calc_rewards
 
@@ -113,7 +106,7 @@ async def resync_metagraph(self):
         self.save_state()
 
     def query_miners(self):
-        timestamp = datetime_to_iso8601(get_now())
+        timestamp = get_str()
         synapse = Challenge(timestamp=timestamp)
         responses = self.dendrite.query(
             # Send the query to selected miner axons in the network.
@@ -163,8 +156,8 @@ async def set_weights(self):
 
     async def scheduled_prediction_request(self):
         if not hasattr(self, "timestamp"):
-            self.timestamp = datetime_to_iso8601(get_before(minutes=self.prediction_interval))
-        query_lag = elapsed_seconds(get_now(), iso8601_to_datetime(self.timestamp))
+            self.timestamp = to_str(get_before(minutes=self.prediction_interval))
+        query_lag = elapsed_seconds(get_now(), to_datetime(self.timestamp))
         if len(self.available_uids) == 0:
             bt.logging.info("No miners available. Sleeping for 10 minutes...")
             print_info(self)

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