130 lines
4.5 KiB
Python
130 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify tick caching with timestamp serialization
|
|
"""
|
|
|
|
import logging
|
|
import sys
|
|
import os
|
|
import pandas as pd
|
|
from datetime import datetime
|
|
|
|
# Add the project root to the path
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from dataprovider_realtime import TickStorage
|
|
|
|
# Set up logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def test_tick_caching():
|
|
"""Test tick caching with pandas Timestamps"""
|
|
logger.info("Testing tick caching with timestamp serialization...")
|
|
|
|
try:
|
|
# Create tick storage
|
|
tick_storage = TickStorage("TEST/SYMBOL", ["1s", "1m"])
|
|
|
|
# Clear any existing cache
|
|
if os.path.exists(tick_storage.cache_path):
|
|
os.remove(tick_storage.cache_path)
|
|
logger.info("Cleared existing cache file")
|
|
|
|
# Add some test ticks with different timestamp formats
|
|
test_ticks = [
|
|
{
|
|
'price': 100.0,
|
|
'quantity': 1.0,
|
|
'timestamp': pd.Timestamp.now()
|
|
},
|
|
{
|
|
'price': 101.0,
|
|
'quantity': 1.5,
|
|
'timestamp': datetime.now()
|
|
},
|
|
{
|
|
'price': 102.0,
|
|
'quantity': 2.0,
|
|
'timestamp': int(datetime.now().timestamp() * 1000) # milliseconds
|
|
}
|
|
]
|
|
|
|
# Add ticks
|
|
for i, tick in enumerate(test_ticks):
|
|
logger.info(f"Adding tick {i+1}: price=${tick['price']}, timestamp type={type(tick['timestamp'])}")
|
|
tick_storage.add_tick(tick)
|
|
|
|
logger.info(f"Total ticks in storage: {len(tick_storage.ticks)}")
|
|
|
|
# Force save to cache
|
|
tick_storage._save_to_cache()
|
|
logger.info("Saved ticks to cache")
|
|
|
|
# Verify cache file exists
|
|
if os.path.exists(tick_storage.cache_path):
|
|
logger.info(f"✅ Cache file created: {tick_storage.cache_path}")
|
|
|
|
# Check file content
|
|
with open(tick_storage.cache_path, 'r') as f:
|
|
import json
|
|
cache_content = json.load(f)
|
|
logger.info(f"Cache contains {len(cache_content)} ticks")
|
|
|
|
# Show first tick to verify format
|
|
if cache_content:
|
|
first_tick = cache_content[0]
|
|
logger.info(f"First tick in cache: {first_tick}")
|
|
logger.info(f"Timestamp type in cache: {type(first_tick['timestamp'])}")
|
|
else:
|
|
logger.error("❌ Cache file was not created")
|
|
return False
|
|
|
|
# Create new tick storage instance to test loading
|
|
logger.info("Creating new TickStorage instance to test loading...")
|
|
new_tick_storage = TickStorage("TEST/SYMBOL", ["1s", "1m"])
|
|
|
|
# Load from cache
|
|
cache_loaded = new_tick_storage._load_from_cache()
|
|
|
|
if cache_loaded:
|
|
logger.info(f"✅ Successfully loaded {len(new_tick_storage.ticks)} ticks from cache")
|
|
|
|
# Verify timestamps are properly converted back to pandas Timestamps
|
|
for i, tick in enumerate(new_tick_storage.ticks):
|
|
logger.info(f"Loaded tick {i+1}: price=${tick['price']}, timestamp={tick['timestamp']}, type={type(tick['timestamp'])}")
|
|
|
|
if not isinstance(tick['timestamp'], pd.Timestamp):
|
|
logger.error(f"❌ Timestamp not properly converted back to pandas.Timestamp: {type(tick['timestamp'])}")
|
|
return False
|
|
|
|
logger.info("✅ All timestamps properly converted back to pandas.Timestamp")
|
|
return True
|
|
else:
|
|
logger.error("❌ Failed to load ticks from cache")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Error in tick caching test: {str(e)}")
|
|
import traceback
|
|
logger.error(traceback.format_exc())
|
|
return False
|
|
|
|
def main():
|
|
"""Run the test"""
|
|
logger.info("🧪 Starting tick caching test...")
|
|
logger.info("=" * 50)
|
|
|
|
success = test_tick_caching()
|
|
|
|
logger.info("\n" + "=" * 50)
|
|
if success:
|
|
logger.info("🎉 Tick caching test PASSED!")
|
|
else:
|
|
logger.error("❌ Tick caching test FAILED!")
|
|
|
|
return success
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
sys.exit(0 if success else 1) |