# import sys
# sys.path.insert(0, '/var/www/html/backtestingmarket')
# import os
# import re
# import hmac
# import hashlib
# import json
# import time
# import requests
# import pandas as pd
# import pandas_market_calendars as mcal
# from datetime import datetime, timezone, timedelta
# from zoneinfo import ZoneInfo
# from celery import shared_task
# BASE_PATH = "/var/www/html/backtestingmarket/predictor_data/data"
# DOLLAR_SYMBOLS = {"SPX", "RUT", "XSP"}
# RISK_COLUMN_MAP = {
# "VERTICAL": {
# "conservador": "IDEA",
# "agresivo": "IDEA_adj5",
# "intermedio": "IDEA_adj10",
# "ultra_agresivo": "IDEA_adj15",
# },
# "IRON_CONDOR": {
# "conservador": "IDEA_IC",
# "agresivo": "IDEA_IC_adj5",
# "intermedio": "IDEA_IC_adj10",
# "ultra_agresivo": "IDEA_IC_adj15",
# },
# }
# def get_csv_path(symbol: str, date: datetime) -> str:
# date_str = date.strftime("%Y-%m-%d")
# prefix = f"${symbol}" if symbol in DOLLAR_SYMBOLS else symbol
# return os.path.join(BASE_PATH, symbol, f"prediction_{prefix}_{date_str}.csv")
# def extract_credit(idea_text: str) -> float | None:
# if not idea_text or pd.isna(idea_text):
# return None
# match = re.search(r'@([\d.]+)\s+LMT', str(idea_text))
# return float(match.group(1)) if match else None
# def get_latest_5min_row(df: pd.DataFrame, now_ny: datetime) -> pd.Series | None:
# df['timestamp'] = pd.to_datetime(df['timestamp'])
# minute = (now_ny.minute // 5) * 5
# target = now_ny.replace(minute=minute, second=0, microsecond=0)
# target_naive = target.replace(tzinfo=None)
# mask = (df['timestamp'] >= target_naive - timedelta(seconds=90)) & \
# (df['timestamp'] <= target_naive + timedelta(seconds=90))
# subset = df[mask]
# if subset.empty:
# return None
# return subset.iloc[-1]
# def sign_payload(payload: dict, secret: str) -> str:
# body = json.dumps(payload, separators=(',', ':')).encode()
# return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
# def _is_market_hours(now_ny: datetime) -> bool:
# # Fines de semana
# if now_ny.weekday() >= 5:
# return False
# # Feriados NYSE
# nyse = mcal.get_calendar('NYSE')
# today_str = now_ny.strftime("%Y-%m-%d")
# schedule = nyse.schedule(start_date=today_str, end_date=today_str)
# if schedule.empty:
# return False # feriado
# market_open = now_ny.replace(hour=10, minute=5, second=0, microsecond=0)
# market_close = now_ny.replace(hour=16, minute=0, second=0, microsecond=0)
# return market_open <= now_ny <= market_close
# @shared_task(name="webhook_task.check_and_fire_webhooks")
# def check_and_fire_webhooks():
# from app import app
# from models import db, UserWebhookRule
# # ── Esperar hasta el segundo :45 del minuto ──────────────
# now_check = datetime.now(ZoneInfo("America/New_York"))
# if now_check.second < 45:
# time.sleep(45 - now_check.second)
# with app.app_context():
# now_ny = datetime.now(ZoneInfo("America/New_York"))
# today = now_ny.date()
# if not _is_market_hours(now_ny):
# return {"status": "skipped", "reason": "outside_market_hours"}
# rules = (
# UserWebhookRule.query
# .join(UserWebhookRule.user)
# .filter(
# UserWebhookRule.active == True,
# UserWebhookRule.user.has(has_webhook_addon=True)
# )
# .all()
# )
# if not rules:
# return {"status": "ok", "fired": 0, "reason": "no_active_rules"}
# csv_cache = {}
# fired = 0
# errors = []
# for rule in rules:
# try:
# result = _evaluate_rule(rule, now_ny, today, csv_cache)
# if result == "fired":
# fired += 1
# rule.last_fired_at = datetime.now(timezone.utc)
# except Exception as e:
# errors.append(f"rule_id={rule.id} error={str(e)}")
# db.session.commit()
# return {
# "status": "ok",
# "fired": fired,
# "evaluated": len(rules),
# "errors": errors,
# "timestamp_ny": now_ny.isoformat(),
# }
# def _evaluate_rule(rule, now_ny: datetime, today, csv_cache: dict) -> str:
# # ── 1. Evitar doble disparo ──
# if rule.last_fired_at:
# last = rule.last_fired_at.replace(tzinfo=timezone.utc).astimezone(ZoneInfo("America/New_York"))
# minutes_since = (now_ny - last).total_seconds() / 60
# if minutes_since < 4:
# return "skipped"
# # ── 2. Verificar hora exacta ──
# if rule.time_from and rule.time_from != "btm":
# current_time = now_ny.strftime("%H:%M")
# if current_time != rule.time_from:
# return "no_match"
# # ── 3. Leer CSV ──
# cache_key = f"{rule.symbol}_{today}"
# if cache_key not in csv_cache:
# csv_path = get_csv_path(rule.symbol, now_ny)
# if not os.path.exists(csv_path):
# return "no_match"
# csv_cache[cache_key] = pd.read_csv(csv_path)
# df = csv_cache[cache_key]
# # ── 4. Obtener fila del horario elegido ──
# row = get_latest_5min_row(df, now_ny)
# if row is None:
# return "no_match"
# # ── 5. Obtener columna según estrategia + riesgo ──
# col = RISK_COLUMN_MAP.get(rule.strategy, {}).get(rule.risk_level)
# if not col or col not in row.index:
# return "no_match"
# idea_text = row[col]
# if not idea_text or pd.isna(idea_text):
# return "no_match"
# # ── 6. Determinar limit_price ──
# market_credit = extract_credit(idea_text)
# limit_price = rule.limit_price if rule.limit_price else market_credit
# # ── 7. Extraer vencimiento ──
# exp_match = re.search(r'(\d{2}\s+\w+\s+\d{2})', str(idea_text))
# expiration = exp_match.group(1) if exp_match else ""
# # ── 8. Construir payload según estrategia ──
# payload = {
# "event": "trade_signal",
# "symbol": rule.symbol,
# "strategy": rule.strategy,
# "side": "SELL",
# "expiration": expiration,
# "limit_price": limit_price,
# "quantity": rule.quantity or 1,
# "signal_time": now_ny.strftime("%H:%M:%S"),
# "date": str(today),
# "fired_at": datetime.now(timezone.utc).isoformat(),
# "rule_id": rule.id,
# }
# if rule.strategy == "VERTICAL":
# strikes_match = re.search(r'(\d+)/(\d+)\s+(CALL|PUT)', str(idea_text))
# if strikes_match:
# sell_strike = int(strikes_match.group(1))
# buy_strike = int(strikes_match.group(2))
# opt_type = strikes_match.group(3)
# payload["legs"] = [
# {"action": "SELL", "strike": sell_strike, "option_type": opt_type},
# {"action": "BUY", "strike": buy_strike, "option_type": opt_type},
# ]
# else:
# payload["legs"] = []
# else: # IRON_CONDOR
# calls_match = re.search(r'CALLS\s+(\d+)/(\d+)', str(idea_text))
# puts_match = re.search(r'PUTS\s+(\d+)/(\d+)', str(idea_text))
# legs = []
# if calls_match:
# legs.append({"action": "SELL", "strike": int(calls_match.group(1)), "option_type": "CALL"})
# legs.append({"action": "BUY", "strike": int(calls_match.group(2)), "option_type": "CALL"})
# if puts_match:
# legs.append({"action": "SELL", "strike": int(puts_match.group(1)), "option_type": "PUT"})
# legs.append({"action": "BUY", "strike": int(puts_match.group(2)), "option_type": "PUT"})
# payload["legs"] = legs
# # ── 9. Disparar webhook ──
# headers = {"Content-Type": "application/json"}
# if rule.webhook_secret:
# headers["X-BTM-Signature"] = sign_payload(payload, rule.webhook_secret)
# response = requests.post(rule.webhook_url, json=payload, headers=headers, timeout=8)
# response.raise_for_status()
# return "fired"
from utils.schwab_client import send_order_to_schwab
from celery import shared_task
from zoneinfo import ZoneInfo
from datetime import datetime, timezone, timedelta
import pandas_market_calendars as mcal
import pandas as pd
import requests
import time
import json
import hashlib
import hmac
import re
import os
import sys
sys.path.insert(0, '/var/www/html/backtestingmarket')
BASE_PATH = "/var/www/html/backtestingmarket/predictor_data/data"
DOLLAR_SYMBOLS = {"SPX", "RUT", "XSP"}
RISK_COLUMN_MAP = {
"VERTICAL": {
"conservador": "IDEA",
"intermedio": "IDEA_adj5",
"agresivo": "IDEA_adj10",
"ultra_agresivo": "IDEA_adj15",
},
"IRON_CONDOR": {
"conservador": "IDEA_IC",
"intermedio": "IDEA_IC_adj5",
"agresivo": "IDEA_IC_adj10",
"ultra_agresivo": "IDEA_IC_adj15",
},
}
SCHWAB_ACCOUNT_HASH = os.environ.get("SCHWAB_ACCOUNT_HASH")
def get_csv_path(symbol: str, date: datetime) -> str:
date_str = date.strftime("%Y-%m-%d")
prefix = f"${symbol}" if symbol in DOLLAR_SYMBOLS else symbol
return os.path.join(BASE_PATH, symbol, f"prediction_{prefix}_{date_str}.csv")
def extract_credit(idea_text: str) -> float | None:
if not idea_text or pd.isna(idea_text):
return None
match = re.search(r'@([\d.]+)\s+LMT', str(idea_text))
return float(match.group(1)) if match else None
def get_latest_5min_row(df: pd.DataFrame, now_ny: datetime) -> pd.Series | None:
df['timestamp'] = pd.to_datetime(df['timestamp'])
minute = (now_ny.minute // 5) * 5
target = now_ny.replace(minute=minute, second=0, microsecond=0)
target_naive = target.replace(tzinfo=None)
mask = (df['timestamp'] >= target_naive - timedelta(seconds=90)) & \
(df['timestamp'] <= target_naive + timedelta(seconds=90))
subset = df[mask]
if subset.empty:
return None
return subset.iloc[-1]
def sign_payload(payload: dict, secret: str) -> str:
body = json.dumps(payload, separators=(',', ':')).encode()
return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
def _is_market_hours(now_ny: datetime) -> bool:
# Fines de semana
if now_ny.weekday() >= 5:
return False
# Feriados NYSE
nyse = mcal.get_calendar('NYSE')
today_str = now_ny.strftime("%Y-%m-%d")
schedule = nyse.schedule(start_date=today_str, end_date=today_str)
if schedule.empty:
return False # feriado
market_open = now_ny.replace(hour=10, minute=5, second=0, microsecond=0)
market_close = now_ny.replace(hour=16, minute=0, second=0, microsecond=0)
return market_open <= now_ny <= market_close
@shared_task(name="webhook_task.check_and_fire_webhooks")
def check_and_fire_webhooks():
from app import app
from models import db, UserWebhookRule
# ── Esperar hasta el segundo :45 del minuto ──────────────
now_check = datetime.now(ZoneInfo("America/New_York"))
if now_check.second < 45:
time.sleep(45 - now_check.second)
with app.app_context():
now_ny = datetime.now(ZoneInfo("America/New_York"))
today = now_ny.date()
if not _is_market_hours(now_ny):
return {"status": "skipped", "reason": "outside_market_hours"}
rules = (
UserWebhookRule.query
.join(UserWebhookRule.user)
.filter(
UserWebhookRule.active == True,
UserWebhookRule.user.has(has_webhook_addon=True)
)
.all()
)
if not rules:
return {"status": "ok", "fired": 0, "reason": "no_active_rules"}
csv_cache = {}
fired = 0
errors = []
for rule in rules:
try:
result = _evaluate_rule(rule, now_ny, today, csv_cache)
if result == "fired":
fired += 1
rule.last_fired_at = datetime.now(timezone.utc)
except Exception as e:
errors.append(f"rule_id={rule.id} error={str(e)}")
db.session.commit()
return {
"status": "ok",
"fired": fired,
"evaluated": len(rules),
"errors": errors,
"timestamp_ny": now_ny.isoformat(),
}
def _evaluate_rule(rule, now_ny: datetime, today, csv_cache: dict) -> str:
# ── 1. Evitar doble disparo ──
if rule.last_fired_at:
last = rule.last_fired_at.replace(tzinfo=timezone.utc).astimezone(ZoneInfo("America/New_York"))
minutes_since = (now_ny - last).total_seconds() / 60
if minutes_since < 4:
return "skipped"
# ── 2. Verificar hora exacta ──
if rule.time_from and rule.time_from != "btm":
current_time = now_ny.strftime("%H:%M")
if current_time != rule.time_from:
return "no_match"
# ── 3. Leer CSV ──
cache_key = f"{rule.symbol}_{today}"
if cache_key not in csv_cache:
csv_path = get_csv_path(rule.symbol, now_ny)
if not os.path.exists(csv_path):
return "no_match"
csv_cache[cache_key] = pd.read_csv(csv_path)
df = csv_cache[cache_key]
# ── 4. Obtener fila del horario elegido ──
row = get_latest_5min_row(df, now_ny)
if row is None:
return "no_match"
# ── 5. Obtener columna según estrategia + riesgo ──
col = RISK_COLUMN_MAP.get(rule.strategy, {}).get(rule.risk_level)
if not col or col not in row.index:
return "no_match"
idea_text = row[col]
if not idea_text or pd.isna(idea_text):
return "no_match"
# ── 6. Determinar limit_price ──
market_credit = extract_credit(idea_text)
limit_price = rule.limit_price if rule.limit_price else market_credit
# ── 7. Extraer vencimiento ──
exp_match = re.search(r'(\d{2}\s+\w+\s+\d{2})', str(idea_text))
expiration = exp_match.group(1) if exp_match else ""
# ── 8. Construir payload según estrategia ──
payload = {
"event": "trade_signal",
"symbol": rule.symbol,
"strategy": rule.strategy,
"side": "SELL",
"expiration": expiration,
"limit_price": limit_price,
"quantity": rule.quantity or 1,
"signal_time": now_ny.strftime("%H:%M:%S"),
"date": str(today),
"fired_at": datetime.now(timezone.utc).isoformat(),
"rule_id": rule.id,
}
if rule.strategy == "VERTICAL":
strikes_match = re.search(r'(\d+)/(\d+)\s+(CALL|PUT)', str(idea_text))
if strikes_match:
sell_strike = int(strikes_match.group(1))
buy_strike = int(strikes_match.group(2))
opt_type = strikes_match.group(3)
payload["legs"] = [
{"action": "SELL", "strike": sell_strike, "option_type": opt_type},
{"action": "BUY", "strike": buy_strike, "option_type": opt_type},
]
else:
payload["legs"] = []
else: # IRON_CONDOR
calls_match = re.search(r'CALLS\s+(\d+)/(\d+)', str(idea_text))
puts_match = re.search(r'PUTS\s+(\d+)/(\d+)', str(idea_text))
legs = []
if calls_match:
legs.append({"action": "SELL", "strike": int(calls_match.group(1)), "option_type": "CALL"})
legs.append({"action": "BUY", "strike": int(calls_match.group(2)), "option_type": "CALL"})
if puts_match:
legs.append({"action": "SELL", "strike": int(puts_match.group(1)), "option_type": "PUT"})
legs.append({"action": "BUY", "strike": int(puts_match.group(2)), "option_type": "PUT"})
payload["legs"] = legs
# ── 9. Disparar: webhook si tiene URL, si no → Schwab ──
if rule.webhook_url:
headers = {"Content-Type": "application/json"}
if rule.webhook_secret:
headers["X-BTM-Signature"] = sign_payload(payload, rule.webhook_secret)
response = requests.post(rule.webhook_url, json=payload, headers=headers, timeout=8)
response.raise_for_status()
else:
send_order_to_schwab(payload, SCHWAB_ACCOUNT_HASH)
return "fired"