implementations

crypto/gogo/data/data_utils.py

@@ -121,6 +121,44 @@ def create_padding_mask(seq, pad_token=0):
     """
     return (seq == pad_token).all(dim=-1).unsqueeze(0)
 
+def get_aligned_candle_with_index(candles_list, base_ts):
+    """
+    Find the candle from candles_list that is closest to (and <=) base_ts.
+    Returns: (index, candle)
+    """
+    aligned_index = None
+    aligned_candle = None
+    for i in range(len(candles_list)):
+        if candles_list[i]["timestamp"] <= base_ts:
+            aligned_index = i
+            aligned_candle = candles_list[i]
+        else:
+            break
+    return aligned_index, aligned_candle
+
+def get_features_for_tf(candles_list, aligned_index, period=10):
+    """
+    Extract features from the candle at aligned_index.
+    If aligned_index is None, return a zeroed feature vector.
+    """
+    if aligned_index is None:
+        return [0.0] * 7  # return zeroed feature vector
+    candle = candles_list[aligned_index]
+    # Simple features: open, high, low, close, volume, and two EMAs.
+    close_prices = [c["close"] for c in candles_list[:aligned_index+1]]
+    ema_short = calculate_ema(candles_list[:aligned_index+1], period=period)[-1]
+    ema_long = calculate_ema(candles_list[:aligned_index+1], period=period*2)[-1]
+    features = [
+        candle["open"],
+        candle["high"],
+        candle["low"],
+        candle["close"],
+        candle["volume"],
+        ema_short,
+        ema_long
+    ]
+    return features
+
 # Example usage (within a larger training loop):
 if __name__ == '__main__':
     # Dummy data for demonstration
@@ -155,4 +193,14 @@ if __name__ == '__main__':
     mask = create_mask(seq_len)
     print("\nMask:\n", mask)
     padding_mask = create_padding_mask(torch.tensor(candle_features))
-    print(f"\nPadding mask: {padding_mask}")
+    print(f"\nPadding mask: {padding_mask}")
+
+    # Example usage of the new functions
+    index, candle = get_aligned_candle_with_index(candles_data, 1678886570000)
+    if candle:
+        print(f"\nAligned candle: {candle}")
+    else:
+        print("\nNo aligned candle found.")
+
+    features = get_features_for_tf(candles_data, index)
+    print(f"\nFeatures for timeframe: {features}")