import requests import os from datetime import date, timedelta from dotenv import load_dotenv from db import get_connection from config import TICKERS from momentum import ( get_historical_prices, get_historical_prices_sp500, calc_ma, calc_rs_vs_sp500, calc_52w_high, calc_volume_trend, get_company_id ) from fundamentals import calc_growth from catalysts import analyze_grades, analyze_insider_buying from scoring import calc_fundamental_score, calc_momentum_score, calc_final_score load_dotenv() API_KEY = os.getenv("FMP_API_KEY") BASE_URL = "https://financialmodelingprep.com/stable" SAMPLE_TICKERS = ["ROK", "VRT", "GXO", "PTC", "POWL"] # ── FETCH: HISTORIAL COMPLETO (una sola llamada por ticker/endpoint) ────── def get_income_statement_history(ticker, period="annual", limit=10): url = f"{BASE_URL}/income-statement?symbol={ticker}&period={period}&limit={limit}&apikey={API_KEY}" r = requests.get(url, timeout=10) return r.json() if r.status_code == 200 else [] def get_cash_flow_history(ticker, period="annual", limit=10): url = f"{BASE_URL}/cash-flow-statement?symbol={ticker}&period={period}&limit={limit}&apikey={API_KEY}" r = requests.get(url, timeout=10) return r.json() if r.status_code == 200 else [] def get_key_metrics_history(ticker, limit=10): """key-metrics solo soporta anual en el plan Starter (period=quarter da HTTP 402).""" url = f"{BASE_URL}/key-metrics?symbol={ticker}&period=annual&limit={limit}&apikey={API_KEY}" r = requests.get(url, timeout=10) return r.json() if r.status_code == 200 else [] def get_grades_history(ticker, limit=500): url = f"{BASE_URL}/grades?symbol={ticker}&limit={limit}&apikey={API_KEY}" r = requests.get(url, timeout=10) return r.json() if r.status_code == 200 else [] def get_insider_trades_history(ticker, limit=500): url = f"{BASE_URL}/insider-trading/search?symbol={ticker}&limit={limit}&apikey={API_KEY}" r = requests.get(url, timeout=10) return r.json() if r.status_code == 200 else [] # ── FECHAS DE SNAPSHOT ─────────────────────────────────────────────────── def get_weekly_snapshot_dates(weeks=208, lookback_buffer_weeks=12, end_date=None): if end_date is None: end_date = date.today() last_valid = end_date - timedelta(weeks=lookback_buffer_weeks) return [last_valid - timedelta(weeks=i) for i in range(weeks - 1, -1, -1)] # ── SLICING POINT-IN-TIME: PRECIOS ────────────────────────────────────── def slice_prices_as_of(prices, as_of_date): as_of_str = as_of_date.isoformat() for i, p in enumerate(prices): if p["date"] <= as_of_str: return prices[i:] return [] def find_price_near_date(prices, target_date): target_str = target_date.isoformat() for p in prices: if p["date"] <= target_str: return p["close"] return None def calc_momentum_as_of(ticker_prices_full, sp500_prices_full, as_of_date): sliced_ticker = slice_prices_as_of(ticker_prices_full, as_of_date) sliced_sp500 = slice_prices_as_of(sp500_prices_full, as_of_date) if not sliced_ticker: return None return { "price": sliced_ticker[0]["close"], "ma50": calc_ma(sliced_ticker, 50), "ma200": calc_ma(sliced_ticker, 200), "rs_vs_sp500_3m": calc_rs_vs_sp500(sliced_ticker, sliced_sp500), "week_high_52": calc_52w_high(sliced_ticker), "volume_trend": calc_volume_trend(sliced_ticker), } # ── SELECCION POINT-IN-TIME: FUNDAMENTALES ────────────────────────────── def attach_filing_dates(metrics_hist, income_hist_annual): filing_by_date = {r["date"]: r.get("filingDate") for r in income_hist_annual} for m in metrics_hist: m["filingDate"] = filing_by_date.get(m["date"], "9999-99-99") return metrics_hist def select_yoy_pair(reports, as_of_date, period="annual"): as_of_str = as_of_date.isoformat() known = [r for r in reports if r.get("filingDate", "9999-99-99") <= as_of_str] offset = 4 if period == "quarter" else 1 if len(known) <= offset: return (known[0] if known else None), None return known[0], known[offset] def calc_fundamentals_as_of(income_hist, cashflow_hist, metrics_hist, income_hist_annual, as_of_date, period="annual"): metrics_hist = attach_filing_dates(metrics_hist, income_hist_annual) cur_income, cmp_income = select_yoy_pair(income_hist, as_of_date, period) cur_cashflow, cmp_cashflow = select_yoy_pair(cashflow_hist, as_of_date, period) cur_metrics, _ = select_yoy_pair(metrics_hist, as_of_date, period="annual") if not cur_income: return None revenue_growth = calc_growth(cur_income["revenue"], cmp_income["revenue"]) if cmp_income else None eps_growth = calc_growth(cur_income["epsDiluted"], cmp_income["epsDiluted"]) if cmp_income else None operating_margin = round(cur_income["operatingIncome"] / cur_income["revenue"] * 100, 2) \ if cur_income.get("operatingIncome") and cur_income.get("revenue") else 0 fcf_growth = None if cur_cashflow and cmp_cashflow: fcf_growth = calc_growth(cur_cashflow["freeCashFlow"], cmp_cashflow["freeCashFlow"]) roic = None net_debt_ebitda = None if cur_metrics: roic = round(cur_metrics.get("returnOnInvestedCapital", 0) * 100, 2) if cur_metrics.get("returnOnInvestedCapital") else None net_debt_ebitda = round(cur_metrics.get("netDebtToEBITDA", 0), 2) if cur_metrics.get("netDebtToEBITDA") else None return { "revenue_growth_yoy": revenue_growth, "eps_growth_yoy": eps_growth, "fcf_growth_yoy": fcf_growth, "roic": roic, "operating_margin": operating_margin, "net_debt_ebitda": net_debt_ebitda, } # ── SELECCION POINT-IN-TIME: CATALIZADORES ────────────────────────────── def calc_catalyst_score_as_of(grades_hist, insider_hist, as_of_date, lookback_days=90): cutoff_str = (as_of_date - timedelta(days=lookback_days)).isoformat() as_of_str = as_of_date.isoformat() grades_window = [g for g in grades_hist if cutoff_str <= g["date"] <= as_of_str] insider_window = [t for t in insider_hist if cutoff_str <= t["transactionDate"] <= as_of_str] upgrades, downgrades = analyze_grades(grades_window) buy_count, sell_count, buy_value, sell_value = analyze_insider_buying(insider_window) pts = 0 if upgrades > 0: pts += 20 if buy_count > 0 and buy_value > sell_value: pts += 20 return round((pts / 40) * 100, 2) # ── GUARDAR RESULTADO (reutiliza cursor, no abre conexion por fila) ───── def save_backtest_result(cursor, company_id, snapshot_date, period, f, m, c, final, price, p4w, p12w, r4w, r12w): cursor.execute(""" INSERT INTO backtest_results (company_id, snapshot_date, fundamentals_period, fundamental_score, momentum_score, catalyst_score, final_score, price_at_snapshot, price_4w_fwd, price_12w_fwd, return_4w_pct, return_12w_pct) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE fundamental_score=%s, momentum_score=%s, catalyst_score=%s, final_score=%s, price_at_snapshot=%s, price_4w_fwd=%s, price_12w_fwd=%s, return_4w_pct=%s, return_12w_pct=%s """, ( company_id, snapshot_date, period, f, m, c, final, price, p4w, p12w, r4w, r12w, f, m, c, final, price, p4w, p12w, r4w, r12w )) # ── LOOP PRINCIPAL ──────────────────────────────────────────────────────── def run_backtest(tickers, weeks=208): snapshots = get_weekly_snapshot_dates(weeks=weeks) sp500 = get_historical_prices_sp500(limit=1300) total_filas = 0 conn = get_connection() cursor = conn.cursor() for ticker in tickers: print(f"Procesando {ticker}...") company_id = get_company_id(ticker) if not company_id: print(f" {ticker}: company_id no encontrado, se omite") continue prices = get_historical_prices(ticker, limit=1300) if not prices: print(f" {ticker}: sin precios, se omite") continue grades_hist = get_grades_history(ticker) insider_hist = get_insider_trades_history(ticker) income_hist_annual = get_income_statement_history(ticker, period="annual", limit=10) income_hist_quarter = get_income_statement_history(ticker, period="quarter", limit=28) cashflow_hist_annual = get_cash_flow_history(ticker, period="annual", limit=10) cashflow_hist_quarter = get_cash_flow_history(ticker, period="quarter", limit=28) metrics_hist = get_key_metrics_history(ticker, limit=10) filas_ticker = 0 for snapshot in snapshots: momentum_data = calc_momentum_as_of(prices, sp500, snapshot) if momentum_data is None: continue momentum_score = calc_momentum_score(momentum_data) catalyst_score = calc_catalyst_score_as_of(grades_hist, insider_hist, snapshot) price_at_snapshot = momentum_data["price"] price_4w = find_price_near_date(prices, snapshot + timedelta(weeks=4)) price_12w = find_price_near_date(prices, snapshot + timedelta(weeks=12)) return_4w = round((price_4w - price_at_snapshot) / price_at_snapshot * 100, 2) if price_4w else None return_12w = round((price_12w - price_at_snapshot) / price_at_snapshot * 100, 2) if price_12w else None for period, income_hist, cashflow_hist in [ ("annual", income_hist_annual, cashflow_hist_annual), ("quarter", income_hist_quarter, cashflow_hist_quarter), ]: fund_data = calc_fundamentals_as_of(income_hist, cashflow_hist, metrics_hist, income_hist_annual, snapshot, period=period) if fund_data is None: continue fund_score = calc_fundamental_score(fund_data) final = calc_final_score(fund_score, momentum_score, catalyst_score) save_backtest_result(cursor, company_id, snapshot, period, fund_score, momentum_score, catalyst_score, final, price_at_snapshot, price_4w, price_12w, return_4w, return_12w) filas_ticker += 1 conn.commit() print(f" {ticker}: {filas_ticker} filas guardadas") total_filas += filas_ticker cursor.close() conn.close() print(f"\nBacktest completado. Total de filas: {total_filas}") if __name__ == "__main__": run_backtest(tickers=TICKERS, weeks=208)