from datetime import date, timedelta from db import get_connection from config import TICKERS from momentum import get_historical_prices, get_company_id STOP_LOSS_PCT = 0.90 PARTIAL_PROFIT_PCT = 1.30 PARTIAL_PROFIT_FRACTION = 0.30 FULL_EXIT_PROFIT_PCT = 1.50 MAX_HOLDING_DAYS = 182 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)] def slice_prices_forward(prices_desc, from_date): from_str = from_date.isoformat() forward = [p for p in prices_desc if p["date"] >= from_str] forward.sort(key=lambda p: p["date"]) return forward def find_price_on_date(prices_desc, target_date): target_str = target_date.isoformat() for p in prices_desc: if p["date"] <= target_str: return p["close"] return None def build_result(entry_price, exit_price, exit_date, reason, partial_taken, entry_date): if partial_taken: ret = PARTIAL_PROFIT_FRACTION * (PARTIAL_PROFIT_PCT - 1) + (1 - PARTIAL_PROFIT_FRACTION) * (exit_price / entry_price - 1) else: ret = (exit_price / entry_price - 1) return { "exit_date": exit_date, "exit_price": round(exit_price, 2), "exit_reason": reason, "partial_taken": partial_taken, "holding_days": (exit_date - entry_date).days, "return_pct": round(ret * 100, 2), } def simulate_trade(prices_asc, entry_date, entry_price): stop_price = entry_price * STOP_LOSS_PCT partial_price = entry_price * PARTIAL_PROFIT_PCT full_exit_price = entry_price * FULL_EXIT_PROFIT_PCT max_exit_date = entry_date + timedelta(days=MAX_HOLDING_DAYS) partial_taken = False for day in prices_asc: day_date = date.fromisoformat(day["date"]) if day_date <= entry_date: continue if day.get("low") is not None and day["low"] <= stop_price: return build_result(entry_price, stop_price, day_date, "STOP_LOSS", partial_taken, entry_date) if day.get("high") is not None and day["high"] >= full_exit_price: partial_taken = True return build_result(entry_price, full_exit_price, day_date, "SALIDA_TOTAL_GANANCIAS", partial_taken, entry_date) if not partial_taken and day.get("high") is not None and day["high"] >= partial_price: partial_taken = True if day_date >= max_exit_date: return build_result(entry_price, day["close"], day_date, "FIN_HOLDING_PERIOD", partial_taken, entry_date) last = prices_asc[-1] return build_result(entry_price, last["close"], date.fromisoformat(last["date"]), "SIN_DATOS_SUFICIENTES", partial_taken, entry_date) def save_trade_result(cursor, company_id, period, entry_date, entry_price, entry_score, result): cursor.execute(""" INSERT INTO trade_simulation_results (company_id, fundamentals_period, entry_date, entry_price, entry_score, exit_date, exit_price, exit_reason, partial_profit_taken, holding_days, return_pct) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE entry_score=%s, exit_date=%s, exit_price=%s, exit_reason=%s, partial_profit_taken=%s, holding_days=%s, return_pct=%s """, ( company_id, period, entry_date, entry_price, entry_score, result["exit_date"], result["exit_price"], result["exit_reason"], result["partial_taken"], result["holding_days"], result["return_pct"], entry_score, result["exit_date"], result["exit_price"], result["exit_reason"], result["partial_taken"], result["holding_days"], result["return_pct"] )) def run_trade_simulation(tickers, periods=("annual", "quarter"), min_score=None): """ min_score=None: modo exploratorio, entra en CADA señal semanal sin filtrar por umbral, guardando el score de entrada para poder calibrar despues. """ conn = get_connection() cursor = conn.cursor(dictionary=True) insert_cursor = conn.cursor() insert_cursor.execute("DELETE FROM trade_simulation_results WHERE fundamentals_period IN ('annual','quarter')") conn.commit() total_trades = 0 for ticker in tickers: company_id = get_company_id(ticker) if not company_id: continue prices = get_historical_prices(ticker, limit=1300) if not prices: continue for period in periods: cursor.execute(""" SELECT snapshot_date, final_score FROM backtest_results WHERE company_id=%s AND fundamentals_period=%s ORDER BY snapshot_date ASC """, (company_id, period)) signals = cursor.fetchall() in_position = False position_exit_date = None trades_ticker = 0 for row in signals: snap_date = row["snapshot_date"] score = float(row["final_score"]) if row["final_score"] is not None else None if in_position: if snap_date <= position_exit_date: continue in_position = False if score is None: continue if min_score is not None and score <= min_score: continue entry_price = find_price_on_date(prices, snap_date) if entry_price is None: continue forward = slice_prices_forward(prices, snap_date) if not forward: continue result = simulate_trade(forward, snap_date, entry_price) save_trade_result(insert_cursor, company_id, period, snap_date, entry_price, score, result) conn.commit() in_position = True position_exit_date = result["exit_date"] trades_ticker += 1 total_trades += 1 if trades_ticker: print(f"{ticker} [{period}]: {trades_ticker} operaciones simuladas") cursor.close() insert_cursor.close() conn.close() print(f"\nTotal operaciones simuladas: {total_trades}") def run_baseline_simulation(tickers, start_date=None): if start_date is None: start_date = get_weekly_snapshot_dates(weeks=208)[0] conn = get_connection() insert_cursor = conn.cursor() insert_cursor.execute("DELETE FROM trade_simulation_results WHERE fundamentals_period='baseline'") conn.commit() total_trades = 0 for ticker in tickers: company_id = get_company_id(ticker) if not company_id: continue prices = get_historical_prices(ticker, limit=1300) if not prices: continue current_date = start_date trades_ticker = 0 while True: entry_price = find_price_on_date(prices, current_date) if entry_price is None: break forward = slice_prices_forward(prices, current_date) if not forward: break result = simulate_trade(forward, current_date, entry_price) save_trade_result(insert_cursor, company_id, "baseline", current_date, entry_price, None, result) conn.commit() trades_ticker += 1 total_trades += 1 if result["exit_reason"] == "SIN_DATOS_SUFICIENTES": break current_date = result["exit_date"] + timedelta(days=1) if trades_ticker: print(f"{ticker} [baseline]: {trades_ticker} operaciones") insert_cursor.close() conn.close() print(f"\nTotal operaciones baseline: {total_trades}") if __name__ == "__main__": run_trade_simulation(tickers=TICKERS, min_score=None) run_baseline_simulation(tickers=TICKERS)