82 lines
1.9 KiB
Python
82 lines
1.9 KiB
Python
"""Sample data generators for testing."""
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime, timedelta
|
|
from faker import Faker
|
|
|
|
fake = Faker()
|
|
|
|
|
|
def generate_ohlcv_data(
|
|
symbol: str = "BTC/USD",
|
|
periods: int = 100,
|
|
start_price: float = 50000.0,
|
|
timeframe: str = "1h"
|
|
) -> pd.DataFrame:
|
|
"""Generate sample OHLCV data.
|
|
|
|
Args:
|
|
symbol: Trading pair
|
|
periods: Number of periods
|
|
start_price: Starting price
|
|
timeframe: Data timeframe
|
|
|
|
Returns:
|
|
DataFrame with OHLCV data
|
|
"""
|
|
dates = pd.date_range(
|
|
start=datetime.now() - timedelta(hours=periods),
|
|
periods=periods,
|
|
freq=timeframe
|
|
)
|
|
|
|
# Generate realistic price movement
|
|
prices = [start_price]
|
|
for i in range(1, periods):
|
|
change = np.random.randn() * 0.02 # 2% volatility
|
|
prices.append(prices[-1] * (1 + change))
|
|
|
|
return pd.DataFrame({
|
|
'timestamp': dates,
|
|
'open': prices,
|
|
'high': [p * 1.01 for p in prices],
|
|
'low': [p * 0.99 for p in prices],
|
|
'close': prices,
|
|
'volume': [1000.0 + np.random.randn() * 100 for _ in range(periods)]
|
|
})
|
|
|
|
|
|
def generate_trade_data(
|
|
symbol: str = "BTC/USD",
|
|
count: int = 10
|
|
) -> list:
|
|
"""Generate sample trade data.
|
|
|
|
Args:
|
|
symbol: Trading pair
|
|
count: Number of trades
|
|
|
|
Returns:
|
|
List of trade dictionaries
|
|
"""
|
|
trades = []
|
|
base_price = 50000.0
|
|
|
|
for i in range(count):
|
|
trades.append({
|
|
'order_id': f'trade_{i}',
|
|
'symbol': symbol,
|
|
'side': 'buy' if i % 2 == 0 else 'sell',
|
|
'type': 'market',
|
|
'price': base_price + np.random.randn() * 100,
|
|
'amount': 0.01,
|
|
'cost': 500.0,
|
|
'fee': 0.5,
|
|
'status': 'filled',
|
|
'timestamp': datetime.now() - timedelta(hours=count-i)
|
|
})
|
|
|
|
return trades
|
|
|