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("""
š gate4: Live Gate.io GPT API
Endpoints:
- /balance
- /performance
- /log_trade
- /openapi.yaml
""")
@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()