Spaces:
Sleeping
Sleeping
File size: 5,032 Bytes
b4ba022 | 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 | import json
import os
from fastapi import FastAPI
import uvicorn
from engine.risk_engine import VertexRiskEngine
from engine.semantic import VertexSemanticAgent
from engine.web3_engine import VertexWeb3Engine
app = FastAPI()
# Archivo de persistencia de la watchlist
WATCHLIST_FILE = "app/watchlist.json"
@app.get("/batch_audit")
def run_batch_audit():
"""Audita toda la watchlist y devuelve alertas para n8n."""
if not os.path.exists(WATCHLIST_FILE):
return {"error": "Watchlist file not found", "alerts": []}
try:
with open(WATCHLIST_FILE, "r") as f:
watchlist = json.load(f)
except Exception as e:
return {"error": f"Failed to read watchlist: {e}", "alerts": []}
results = []
for ticker in watchlist:
try:
# Reutilizamos la lógica del motor Némesis
engine = VertexRiskEngine(ticker.upper())
num_res = engine.run_audit()
sem_agent = VertexSemanticAgent(ticker.upper())
risk_text = sem_agent.get_sec_risk_factors()
sem_res = sem_agent.judge_risks(risk_text)
# Blindaje de tipos para evitar errores de comparación
z_score = float(num_res.get('altman_z', 0.0))
m_dsri = float(num_res.get('m_score_dsri', 0.0))
s_score = float(sem_res.get('semantic_score', 0.0))
# Lógica de Semáforo de Riesgo
if z_score < 1.8 or m_dsri > 1.4 or s_score > 3:
status = "RED"
elif z_score < 3.0:
status = "YELLOW"
else:
status = "GREEN"
results.append({
"ticker": ticker,
"status": status,
"z_score": z_score,
"summary": sem_res.get("summary", ""),
"alert": True if status == "RED" else False
})
except Exception as ticker_error:
results.append({"ticker": ticker, "status": "ERROR", "msg": str(ticker_error)})
critical_alerts = [r for r in results if r.get("status") in ["RED", "YELLOW"]]
return {
"total_analyzed": len(watchlist),
"critical_count": len(critical_alerts),
"alerts": critical_alerts,
"full_results": results
}
@app.get("/audit/{ticker}")
def audit_company(ticker: str):
"""Auditoría individual para el tab de Stock Audit."""
try:
engine = VertexRiskEngine(ticker.upper())
num_res = engine.run_audit()
sem_agent = VertexSemanticAgent(ticker.upper())
risk_text = sem_agent.get_sec_risk_factors()
sem_res = sem_agent.judge_risks(risk_text)
# Blindaje de tipos
z_score = float(num_res.get('altman_z', 0.0))
m_dsri = float(num_res.get('m_score_dsri', 0.0))
s_score = float(sem_res.get('semantic_score', 0.0))
if z_score < 1.8 or m_dsri > 1.4 or s_score > 3:
status = "RED"
msg = "CRITICAL RISK: Red flags detected."
elif z_score < 3.0:
status = "YELLOW"
msg = "CAUTION: Monitor closely."
else:
status = "GREEN"
msg = "SAFE: Solid fundamentals."
return {
"ticker": ticker.upper(),
"status": status,
"numeric_analysis": {"altman_z": z_score, "m_score_dsri": m_dsri},
"semantic_analysis": sem_res,
"msg": msg
}
except Exception as e:
return {"status": "ERROR", "msg": str(e)}
@app.get("/audit_contract/{address}")
def audit_smart_contract(address: str):
"""Auditoría de Web3 usando el nuevo motor V2."""
try:
web3_engine = VertexWeb3Engine(address)
audit_res = web3_engine.get_contract_source()
if not audit_res["success"]:
return {"status": "ERROR", "msg": audit_res["error"]}
source = audit_res["source_code"]
vulnerabilities = web3_engine.scan_basic_vulnerabilities(source)
return {
"address": address,
"status": "DANGER" if len(vulnerabilities) > 0 else "SAFE",
"vulnerabilities": vulnerabilities,
"source_preview": source[:500] + "..." # Ahora source es string y el slice no falla
}
except Exception as e:
return {"status": "ERROR", "msg": str(e)}
@app.get("/get_settings")
async def get_settings():
# Esta línea es la que saca las llaves de los Secrets de Hugging Face
token = os.environ.get("BOT_TOKEN", "")
chat = os.environ.get("CHAT_ID", "")
if token and chat:
return {"bot_token": token, "chat_id": chat, "status": "loaded_from_secrets"}
return {"bot_token": "", "chat_id": "", "status": "secrets_not_found"}
if __name__ == "__main__":
# Hugging Face SIEMPRE usa el puerto 7860 internamente
port = int(os.environ.get("PORT", 7860))
# Importante: host="0.0.0.0" para que sea accesible desde fuera del contenedor
uvicorn.run(app, host="0.0.0.0", port=port)
|