125 lines
4.4 KiB
Python
125 lines
4.4 KiB
Python
import logging
|
|
from datetime import timedelta
|
|
from django.utils import timezone
|
|
from django.db.models import Avg, Min, Max
|
|
from decimal import Decimal
|
|
|
|
from monitor.models import BitcoinPrice, MarketAnalysis
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MarketAnalyzer:
|
|
"""Service for analyzing Bitcoin market data."""
|
|
|
|
def __init__(self, threshold_percent=15.0):
|
|
self.threshold_percent = Decimal(str(threshold_percent))
|
|
|
|
def analyze_market(self, period='hourly'):
|
|
"""Analyze the Bitcoin market for a given period."""
|
|
try:
|
|
# Get data for the period
|
|
time_filter = self._get_time_filter(period)
|
|
prices = BitcoinPrice.objects.filter(timestamp__gte=time_filter)
|
|
|
|
if not prices.exists():
|
|
logger.warning(f"No price data available for {period} analysis")
|
|
return None
|
|
|
|
# Get latest price
|
|
latest_price = prices.latest('timestamp')
|
|
|
|
# Calculate statistics
|
|
stats = prices.aggregate(
|
|
avg=Avg('price_usd'),
|
|
min=Min('price_usd'),
|
|
max=Max('price_usd'),
|
|
)
|
|
|
|
avg_price = stats['avg'] or latest_price.price_usd
|
|
min_price = stats['min'] or latest_price.price_usd
|
|
max_price = stats['max'] or latest_price.price_usd
|
|
|
|
# Calculate thresholds
|
|
lower_threshold = avg_price * (1 - self.threshold_percent / 100)
|
|
upper_threshold = avg_price * (1 + self.threshold_percent / 100)
|
|
|
|
# Determine status
|
|
current_price = latest_price.price_usd
|
|
|
|
if current_price < lower_threshold:
|
|
status = 'dip'
|
|
is_event = True
|
|
event_type = 'dip_below'
|
|
elif current_price > upper_threshold:
|
|
status = 'peak'
|
|
is_event = True
|
|
event_type = 'rise_above'
|
|
else:
|
|
status = 'neutral'
|
|
is_event = False
|
|
event_type = ''
|
|
|
|
# Save analysis
|
|
analysis = MarketAnalysis.objects.create(
|
|
period=period,
|
|
current_price=current_price,
|
|
average_price=avg_price,
|
|
min_price=min_price,
|
|
max_price=max_price,
|
|
status=status,
|
|
threshold_percent=self.threshold_percent,
|
|
lower_threshold=lower_threshold,
|
|
upper_threshold=upper_threshold,
|
|
is_event=is_event,
|
|
event_type=event_type,
|
|
)
|
|
|
|
logger.info(f"Market analysis saved: {status} at ${current_price}")
|
|
return analysis
|
|
|
|
except BitcoinPrice.DoesNotExist:
|
|
logger.error("No Bitcoin price data found")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error analyzing market: {e}")
|
|
return None
|
|
|
|
def _get_time_filter(self, period):
|
|
"""Get datetime filter based on period."""
|
|
now = timezone.now()
|
|
|
|
if period == 'hourly':
|
|
return now - timedelta(hours=1)
|
|
elif period == 'daily':
|
|
return now - timedelta(days=1)
|
|
elif period == 'weekly':
|
|
return now - timedelta(weeks=1)
|
|
elif period == 'yearly':
|
|
return now - timedelta(days=365)
|
|
else:
|
|
return now - timedelta(days=1) # Default to daily
|
|
|
|
def get_latest_analysis(self, period='hourly'):
|
|
"""Get the latest analysis for a period."""
|
|
try:
|
|
return MarketAnalysis.objects.filter(period=period).latest('timestamp')
|
|
except MarketAnalysis.DoesNotExist:
|
|
return None
|
|
def get_analysis_summary(self):
|
|
"""Get summary of all analyses."""
|
|
summary = {}
|
|
|
|
for period in ['hourly', 'daily', 'weekly', 'yearly']:
|
|
analysis = self.get_latest_analysis(period)
|
|
if analysis:
|
|
summary[period] = {
|
|
'status': analysis.status,
|
|
'current_price': float(analysis.current_price),
|
|
'average_price': float(analysis.average_price),
|
|
'threshold_percent': float(analysis.threshold_percent),
|
|
'is_event': analysis.is_event,
|
|
}
|
|
|
|
return summary
|