199 lines
7.2 KiB
Python
199 lines
7.2 KiB
Python
import random
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Optional
|
|
from app.schemas.schemas import PriceData
|
|
|
|
|
|
class GoldPriceSimulator:
|
|
"""
|
|
Simulates realistic gold price movements without external API calls.
|
|
Uses Geometric Brownian Motion for realistic price action.
|
|
"""
|
|
|
|
def __init__(self, initial_price: float = 2650.0):
|
|
"""
|
|
Initialize the simulator with a starting price.
|
|
|
|
Args:
|
|
initial_price: Starting gold price in USD per oz (default ~current market price)
|
|
"""
|
|
self.base_price = initial_price
|
|
self.current_price = initial_price
|
|
self.volatility = 0.0008 # Daily volatility (0.08%)
|
|
self.drift = 0.00001 # Slight upward drift
|
|
self.last_update = time.time()
|
|
|
|
# For trend simulation
|
|
self.trend_direction = 1 # 1 for up, -1 for down
|
|
self.trend_strength = 0.0001
|
|
self.trend_duration = 0
|
|
self.max_trend_duration = 100 # Max ticks before trend change
|
|
|
|
def _calculate_price_change(self) -> float:
|
|
"""Calculate the next price change using Geometric Brownian Motion."""
|
|
# Random walk component
|
|
random_shock = random.gauss(0, 1) * self.volatility
|
|
|
|
# Trend component (changes periodically)
|
|
self.trend_duration += 1
|
|
if self.trend_duration > self.max_trend_duration:
|
|
# Change trend direction
|
|
self.trend_direction = random.choice([1, -1])
|
|
self.trend_strength = random.uniform(0.00005, 0.0002)
|
|
self.trend_duration = 0
|
|
self.max_trend_duration = random.randint(50, 200)
|
|
|
|
trend_component = self.trend_direction * self.trend_strength
|
|
|
|
# Mean reversion (pulls price back toward base)
|
|
mean_reversion = (self.base_price - self.current_price) * 0.00001
|
|
|
|
# Combine components
|
|
total_change = self.drift + random_shock + trend_component + mean_reversion
|
|
|
|
return self.current_price * total_change
|
|
|
|
def get_current_price(self) -> float:
|
|
"""Get the current simulated gold price."""
|
|
# Update price based on time elapsed
|
|
current_time = time.time()
|
|
time_elapsed = current_time - self.last_update
|
|
|
|
# Update price (simulating continuous price movement)
|
|
if time_elapsed > 0:
|
|
# Multiple small updates for smoother price action
|
|
updates = max(1, int(time_elapsed))
|
|
for _ in range(min(updates, 10)): # Cap at 10 updates to avoid huge jumps
|
|
price_change = self._calculate_price_change()
|
|
self.current_price += price_change
|
|
|
|
# Keep price within reasonable bounds (±20% from base)
|
|
self.current_price = max(
|
|
self.base_price * 0.8,
|
|
min(self.base_price * 1.2, self.current_price)
|
|
)
|
|
|
|
self.last_update = current_time
|
|
return round(self.current_price, 2)
|
|
|
|
def get_live_candle(self, interval: str = "1min") -> PriceData:
|
|
"""
|
|
Generate a live price candle for the current interval.
|
|
|
|
Args:
|
|
interval: Time interval (1min, 5min, 15min, 30min, 60min)
|
|
|
|
Returns:
|
|
PriceData object with OHLC values
|
|
"""
|
|
current_price = self.get_current_price()
|
|
|
|
# Map intervals to seconds
|
|
interval_map = {
|
|
"1min": 60,
|
|
"5min": 5 * 60,
|
|
"15min": 15 * 60,
|
|
"30min": 30 * 60,
|
|
"60min": 60 * 60,
|
|
}
|
|
|
|
interval_seconds = interval_map.get(interval, 60)
|
|
current_time = int(time.time())
|
|
|
|
# Round up to next interval boundary to ensure newest timestamp
|
|
timestamp = ((current_time // interval_seconds) + 1) * interval_seconds
|
|
|
|
# Generate OHLC with small realistic variance
|
|
variance = current_price * 0.0005 # 0.05% variance
|
|
|
|
open_price = current_price + random.uniform(-variance, variance)
|
|
close_price = current_price + random.uniform(-variance, variance)
|
|
high_price = max(open_price, close_price) + random.uniform(0, variance)
|
|
low_price = min(open_price, close_price) - random.uniform(0, variance)
|
|
|
|
return PriceData(
|
|
time=timestamp,
|
|
open=round(open_price, 2),
|
|
high=round(high_price, 2),
|
|
low=round(low_price, 2),
|
|
close=round(close_price, 2),
|
|
)
|
|
|
|
def generate_historical_data(
|
|
self,
|
|
interval: str = "daily",
|
|
points: int = 100
|
|
) -> List[PriceData]:
|
|
"""
|
|
Generate historical price data using the simulator.
|
|
|
|
Args:
|
|
interval: Time interval (daily, 1min, 5min, etc.)
|
|
points: Number of data points to generate
|
|
|
|
Returns:
|
|
List of PriceData objects in chronological order
|
|
"""
|
|
# Map intervals to seconds
|
|
interval_map = {
|
|
"daily": 24 * 60 * 60,
|
|
"1min": 60,
|
|
"5min": 5 * 60,
|
|
"15min": 15 * 60,
|
|
"30min": 30 * 60,
|
|
"60min": 60 * 60,
|
|
}
|
|
|
|
interval_seconds = interval_map.get(interval, 24 * 60 * 60)
|
|
|
|
# Start from past and work forward
|
|
end_time = int(time.time())
|
|
start_time = end_time - (interval_seconds * points)
|
|
|
|
price_data = []
|
|
current_sim_price = self.base_price
|
|
|
|
for i in range(points):
|
|
timestamp = start_time + (interval_seconds * i)
|
|
|
|
# Simulate price evolution
|
|
price_change = random.gauss(0, 1) * self.volatility * current_sim_price
|
|
trend = random.uniform(-0.0001, 0.0001) * current_sim_price
|
|
current_sim_price += price_change + trend
|
|
|
|
# Keep within bounds
|
|
current_sim_price = max(
|
|
self.base_price * 0.85,
|
|
min(self.base_price * 1.15, current_sim_price)
|
|
)
|
|
|
|
# Generate OHLC for this candle
|
|
candle_variance = current_sim_price * 0.002 # 0.2% intra-candle variance
|
|
|
|
open_price = current_sim_price + random.uniform(-candle_variance/2, candle_variance/2)
|
|
close_price = current_sim_price + random.uniform(-candle_variance/2, candle_variance/2)
|
|
high_price = max(open_price, close_price) + random.uniform(0, candle_variance)
|
|
low_price = min(open_price, close_price) - random.uniform(0, candle_variance)
|
|
|
|
price_data.append(
|
|
PriceData(
|
|
time=timestamp,
|
|
open=round(open_price, 2),
|
|
high=round(high_price, 2),
|
|
low=round(low_price, 2),
|
|
close=round(close_price, 2),
|
|
)
|
|
)
|
|
|
|
# Set current price to the last closing price for continuity
|
|
if price_data:
|
|
self.current_price = price_data[-1].close
|
|
self.last_update = time.time()
|
|
|
|
return price_data
|
|
|
|
|
|
# Global simulator instance (maintains state across requests)
|
|
gold_simulator = GoldPriceSimulator(initial_price=2650.0)
|