Spaces:
Sleeping
Sleeping
File size: 5,469 Bytes
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 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 |
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Dict, Any, List, Optional
# Import core logic modules
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
# Initialize core components
data_processor = DataProcessor()
sentiment_analyzer = SentimentAnalyzer()
model_handler = ModelHandler()
trading_logic = TradingLogic()
# FastAPI app setup
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."
)
# Add CORS middleware for frontend access
# HATI-HATI: Ganti "*" dengan domain frontend React Anda saat deployment produksi
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# --- Skema Respon Pydantic ---
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 # String base64 gambar, siap untuk tag <img src="data:image/png;base64,...">
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]
# --- Endpoint API ---
@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)")
):
"""
Mengambil data pasar, menghitung indikator, menghasilkan prediksi,
dan mengembalikan gambar chart (Base64) serta metrik trading.
"""
try:
# 1. Fetch data
df = data_processor.get_market_data(ticker, interval)
if df.empty:
return ChartAnalysisResponse(error=f"No data available for {ticker} at {interval}")
# 2. Calculate Indicators
df = data_processor.calculate_indicators(df)
# 3. Prepare and Predict
prepared_data = data_processor.prepare_for_chronos(df)
predictions = model_handler.predict(prepared_data, horizon=10)
current_price = df['Close'].iloc[-1]
# 4. Generate Chart (returns Base64 HTML string)
chart_html = create_mplfinance_chart(
df,
ticker=f'{ticker} ({interval})',
predictions=predictions
)
# 5. Generate Signal and Metrics
signal, confidence = trading_logic.generate_signal(
predictions, current_price, df
)
tp, sl = trading_logic.calculate_tp_sl(
current_price, df['ATR'].iloc[-1], signal
)
# 6. Format Metrics
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:
# Gunakan HTTPException untuk penanganan kesalahan API
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)")):
"""
Menganalisis dan mengembalikan skor sentimen pasar dan ringkasan berita (Simulasi).
"""
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)")):
"""
Mengambil data fundamental pasar utama (Simulasi).
"""
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)}") |