Spaces:
Sleeping
Sleeping
File size: 4,730 Bytes
4e0dc71 19698be 4e0dc71 19698be 4e0dc71 19698be 4e0dc71 19698be 4e0dc71 19698be 4e0dc71 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 |
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Dict, Any, List, Optional
from data_processor import DataProcessor
from sentiment_analyzer import SentimentAnalyzer
from model_handler import ModelHandler
from trading_logic import TradingLogic
from plotter import create_mplfinance_chart
data_processor = DataProcessor()
sentiment_analyzer = SentimentAnalyzer()
model_handler = ModelHandler()
trading_logic = TradingLogic()
app = FastAPI(
title="Ultimate Market Analysis & Prediction API",
version="1.0.0",
description="API for fetching market data, technical indicators, Chronos-2 predictions, and simulated analysis for GC=F and BTC-USD."
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
class TradingMetrics(BaseModel):
Ticker: str
Current_Price: str
Signal: str
Confidence: str
Take_Profit: str
Stop_Loss: str
RSI: str
MACD: str
Volume: str
class ChartAnalysisResponse(BaseModel):
chart_html_base64: Optional[str] = None
metrics: Optional[TradingMetrics] = None
raw_predictions: Optional[List[float]] = None
error: Optional[str] = None
class SentimentAnalysisResponse(BaseModel):
sentiment_score: float
news_summary_html: str
class FundamentalsResponse(BaseModel):
fundamentals_data: Dict[str, Any]
@app.get("/")
def read_root():
return {"message": "Welcome to the Ultimate Market Analysis API. Use /docs for API documentation."}
@app.get("/analysis/chart", response_model=ChartAnalysisResponse)
def get_chart_analysis(
ticker: str = Query(..., description="Market Ticker (e.g., GC=F, BTC-USD)"),
interval: str = Query(..., description="Time Interval (e.g., 1d, 1h, 5m)")
):
try:
df = data_processor.get_market_data(ticker, interval)
if df.empty:
return ChartAnalysisResponse(error=f"No data available for {ticker} at {interval}")
df = data_processor.calculate_indicators(df)
prepared_data = data_processor.prepare_for_chronos(df)
predictions = model_handler.predict(prepared_data, horizon=10)
current_price = df['Close'].iloc[-1]
chart_html = create_mplfinance_chart(
df,
ticker=f'{ticker} ({interval})',
predictions=predictions
)
signal, confidence = trading_logic.generate_signal(
predictions, current_price, df
)
# Pastikan ATR tersedia sebelum menghitung TP/SL
atr_value = df['ATR'].iloc[-1] if 'ATR' in df.columns and not df['ATR'].empty else 0
tp, sl = trading_logic.calculate_tp_sl(
current_price, atr_value, signal
)
metrics = TradingMetrics(
Ticker=ticker,
Current_Price=f"${current_price:.2f}",
Signal=signal.upper(),
Confidence=f"{confidence:.1%}",
Take_Profit=f"${tp:.2f}" if tp else "N/A",
Stop_Loss=f"${sl:.2f}" if sl else "N/A",
RSI=f"{df['RSI'].iloc[-1]:.1f}",
MACD=f"{df['MACD'].iloc[-1]:.4f}",
Volume=f"{df['Volume'].iloc[-1]:,.0f}"
)
return ChartAnalysisResponse(
chart_html_base64=chart_html,
metrics=metrics,
raw_predictions=predictions.tolist()
)
except Exception as e:
# PENTING: Menggunakan HTTPException 500 dengan detail spesifik
raise HTTPException(status_code=500, detail=f"Error in chart analysis: {str(e)}")
@app.get("/analysis/sentiment", response_model=SentimentAnalysisResponse)
def get_sentiment_analysis(ticker: str = Query(..., description="Market Ticker (e.g., GC=F, BTC-USD)")):
try:
sentiment_score, news_summary_html = sentiment_analyzer.analyze_market_sentiment(ticker)
return SentimentAnalysisResponse(
sentiment_score=sentiment_score,
news_summary_html=news_summary_html
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error in sentiment analysis: {str(e)}")
@app.get("/analysis/fundamentals", response_model=FundamentalsResponse)
def get_fundamentals_analysis(ticker: str = Query(..., description="Market Ticker (e.g., GC=F, BTC-USD)")):
try:
fundamentals = data_processor.get_fundamental_data(ticker)
return FundamentalsResponse(fundamentals_data=fundamentals)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error in fundamentals analysis: {str(e)}") |