Spaces:
Sleeping
Sleeping
File size: 4,062 Bytes
2a0f014 | 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 | import yfinance as yf
import pandas as pd
from fastapi import FastAPI
import uvicorn
app = FastAPI(title="Vertex Coders Risk API")
class VertexRiskEngine:
def __init__(self, ticker):
self.ticker_str = ticker
self.company = yf.Ticker(ticker)
self.scores = {}
def get_value(self, df, possible_keys):
"""Busca un valor en el DataFrame intentando varias llaves posibles."""
for key in possible_keys:
if key in df.index:
return df.loc[key]
raise ValueError(f"Ninguna de las llaves {possible_keys} encontrada.")
def run_audit(self):
try:
balance = self.company.balance_sheet
financials = self.company.financials
# --- ALTMAN Z-SCORE ---
# Usamos iloc[0] para el año más reciente
working_capital = self.get_value(balance, ['Working Capital']).iloc[0]
total_assets = self.get_value(balance, ['Total Assets']).iloc[0]
retained_earnings = self.get_value(balance, ['Retained Earnings']).iloc[0]
ebit = self.get_value(financials, ['EBIT']).iloc[0]
total_liabilities = self.get_value(balance, ['Total Liabilities Net Minority Interest', 'Total Liabilities']).iloc[0]
sales = self.get_value(financials, ['Total Revenue']).iloc[0]
market_cap = self.company.info.get('marketCap', 1)
z = (1.2 * (working_capital/total_assets) +
1.4 * (retained_earnings/total_assets) +
3.3 * (ebit/total_assets) +
0.6 * (market_cap/total_liabilities) +
1.0 * (sales/total_assets))
self.scores['altman_z'] = round(z, 2)
# --- BENEISH M-SCORE (DSRI) ---
# Comparamos Año T (iloc[0]) vs Año T-1 (iloc[1])
sales_t = sales
sales_t1 = self.get_value(financials, ['Total Revenue']).iloc[1]
receivables_keys = ['Net Receivables', 'Accounts Receivable', 'Receivables']
rec_t = self.get_value(balance, receivables_keys).iloc[0]
rec_t1 = self.get_value(balance, receivables_keys).iloc[1]
dsri = (rec_t / sales_t) / (rec_t1 / sales_t1)
self.scores['m_score_dsri'] = round(dsri, 2)
return self.scores
except Exception as e:
return {"error": str(e)}
@app.get("/audit/{ticker}")
def audit_company(ticker: str):
engine = VertexRiskEngine(ticker.upper())
result = engine.run_audit()
# 1. Validar si el motor devolvió un error de ejecución
if "error" in result:
return {
"ticker": ticker.upper(),
"status": "ERROR_TECNICO",
"analysis": None,
"msg": f"No se pudo completar la auditoría: {result['error']}"
}
# 2. Extraer valores con seguridad (Type Casting)
# Forzamos a float para evitar errores de comparación con tipos desconocidos
try:
z_score = float(result.get('altman_z', 0))
m_dsri = float(result.get('m_score_dsri', 0))
except (ValueError, TypeError):
z_score = 0.0
m_dsri = 0.0
# 3. Lógica de Semáforo Robusta
# Prioridad: Rojo (Peligro inminente o posible fraude)
if z_score < 1.8 or m_dsri > 1.4:
status = "ROJO"
msg = "Riesgo crítico detectado: Posible quiebra o manipulación."
# Zona Gris
elif 1.8 <= z_score < 3.0:
status = "AMARILLO"
msg = "Empresa en zona gris. Requiere supervisión manual."
# Zona Segura
else:
status = "VERDE"
msg = "Fundamentos financieros sólidos según modelos automáticos."
return {
"ticker": ticker.upper(),
"status": status,
"analysis": result,
"msg": msg,
"engine_version": "1.0-Némesis"
}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8010) |