|
|
from fastapi import FastAPI, Request |
|
|
from fastapi.responses import HTMLResponse, FileResponse |
|
|
import os, time, hmac, hashlib, requests, json |
|
|
|
|
|
app = FastAPI() |
|
|
|
|
|
GATE_API_KEY = os.getenv("GATE_API_KEY") |
|
|
GATE_API_SECRET = os.getenv("GATE_API_SECRET") |
|
|
GATE_API_BASE = "https://api.gate.io/api/v4" |
|
|
|
|
|
LOG_FILE = "trading_log.json" |
|
|
BAL_FILE = "balance_snapshots.json" |
|
|
|
|
|
@app.get("/") |
|
|
def home(): |
|
|
return HTMLResponse(""" |
|
|
<html><body> |
|
|
<h2>🚀 gate4: Live Gate.io GPT API</h2> |
|
|
<p>Endpoints:</p> |
|
|
<ul> |
|
|
<li>/balance</li> |
|
|
<li>/performance</li> |
|
|
<li>/log_trade</li> |
|
|
<li>/openapi.yaml</li> |
|
|
</ul> |
|
|
</body></html> |
|
|
""") |
|
|
|
|
|
@app.get("/openapi.yaml") |
|
|
def get_openapi(): |
|
|
return FileResponse("openapi.yaml", media_type="text/yaml") |
|
|
|
|
|
@app.get("/balance") |
|
|
def get_balance(): |
|
|
path = "/futures/usdt/accounts" |
|
|
method = "GET" |
|
|
timestamp = str(int(time.time())) |
|
|
body = "" |
|
|
query = "" |
|
|
sign = sign_request(method, path, query, body, timestamp) |
|
|
|
|
|
headers = { |
|
|
"KEY": GATE_API_KEY, |
|
|
"Timestamp": timestamp, |
|
|
"SIGN": sign |
|
|
} |
|
|
|
|
|
url = f"{GATE_API_BASE}{path}" |
|
|
res = requests.get(url, headers=headers) |
|
|
res.raise_for_status() |
|
|
|
|
|
accounts = res.json() |
|
|
total = sum(float(acc.get("available", 0)) for acc in accounts) |
|
|
|
|
|
log_balance(total) |
|
|
return {"balance": round(total, 2)} |
|
|
|
|
|
@app.get("/performance") |
|
|
def get_performance(): |
|
|
if not os.path.exists(LOG_FILE): |
|
|
return {"summary": "No trades logged yet."} |
|
|
|
|
|
with open(LOG_FILE, "r") as f: |
|
|
trades = [json.loads(line) for line in f if line.strip()] |
|
|
|
|
|
total_pnl = sum(t.get("pnl_estimate", 0) for t in trades) |
|
|
summary = f"📊 Total PnL: ${total_pnl:.2f} from {len(trades)} trades." |
|
|
for t in trades[-5:]: |
|
|
summary += f"\n• {t['action']} {t['contract']} → ${t['pnl_estimate']} ({t['reason']})" |
|
|
return {"summary": summary} |
|
|
|
|
|
@app.post("/proxy") |
|
|
async def proxy_gateio(request: Request): |
|
|
payload = await request.json() |
|
|
method = payload.get("method", "GET").upper() |
|
|
path = payload.get("path", "") |
|
|
query = payload.get("query", "") |
|
|
body = payload.get("body", "") |
|
|
timestamp = str(int(time.time())) |
|
|
message = f"{method}\n{path}\n{query}\n{body}\n{timestamp}" |
|
|
signature = hmac.new(api_secret.encode(), message.encode(), hashlib.sha512).hexdigest() |
|
|
|
|
|
headers = { |
|
|
"KEY": api_key, |
|
|
"Timestamp": timestamp, |
|
|
"SIGN": signature, |
|
|
"Content-Type": "application/json" |
|
|
} |
|
|
|
|
|
url = f"https://api.gate.io/api/v4{path}" |
|
|
try: |
|
|
if method == "GET": |
|
|
response = requests.get(url, headers=headers) |
|
|
elif method == "POST": |
|
|
response = requests.post(url, headers=headers, data=body) |
|
|
else: |
|
|
return JSONResponse(status_code=400, content={"error": "Unsupported method"}) |
|
|
|
|
|
return JSONResponse(status_code=response.status_code, content=response.json()) |
|
|
except Exception as e: |
|
|
return {"error": str(e)} |
|
|
|
|
|
|
|
|
@app.post("/log_trade") |
|
|
async def log_trade(request: Request): |
|
|
trade = await request.json() |
|
|
with open(LOG_FILE, "a") as f: |
|
|
f.write(json.dumps(trade) + "\n") |
|
|
return {"status": "logged", "trade": trade} |
|
|
|
|
|
def log_balance(balance): |
|
|
entry = {"timestamp": int(time.time()), "balance": balance} |
|
|
with open(BAL_FILE, "a") as f: |
|
|
f.write(json.dumps(entry) + "\n") |
|
|
|
|
|
def sign_request(method, path, query_string, body, timestamp): |
|
|
message = f"{method}\n{path}\n{query_string}\n{body}\n{timestamp}" |
|
|
return hmac.new(GATE_API_SECRET.encode(), message.encode(), hashlib.sha512).hexdigest() |
|
|
|
