#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ oplab_logger.py ================ Logger de produção: coleta spot + opções (bid/ask/strike/gregas) do BOVA11 via OpLab, a cada 5 min, das 10:15 às 16:45. Feito para rodar via systemd timer (ver oplab-logger.service / oplab-logger.timer nesta mesma pasta). Deploy esperado: /home/jrussi/python/oplab_bova_logger/ oplab_logger.py <- este arquivo .env <- credenciais (NÃO versionar / manter chmod 600) requirements.txt oplab-logger.service oplab-logger.timer data/oplab_options_log_AAAA-MM-DD.csv <- criado automaticamente (1 por dia) logs/oplab_logger.log <- criado automaticamente Uso manual (fora do systemd, para testar): cd /home/jrussi/python/oplab_bova_logger/ source venv/bin/activate python oplab_logger.py --inspect # confere campos da API uma vez python oplab_logger.py --run # roda o loop (ctrl+C para parar) python oplab_logger.py --once # faz UMA coleta e sai (útil p/ teste do systemd) """ from __future__ import annotations import argparse import csv import logging import logging.handlers import os import sys import time from dataclasses import dataclass from datetime import datetime, time as dtime from pathlib import Path from typing import Optional import requests BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" LOG_DIR = BASE_DIR / "logs" DATA_DIR.mkdir(exist_ok=True) LOG_DIR.mkdir(exist_ok=True) # Carrega .env se python-dotenv estiver instalado (opcional, mas recomendado). try: from dotenv import load_dotenv load_dotenv(BASE_DIR / ".env") except ImportError: pass # segue usando variáveis de ambiente já exportadas no shell/systemd # ============================================================================ # LOGGING (console + arquivo rotativo) # ============================================================================ def configurar_logging(): root = logging.getLogger() root.setLevel(logging.INFO) fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s") console = logging.StreamHandler(sys.stdout) console.setFormatter(fmt) root.addHandler(console) arquivo = logging.handlers.RotatingFileHandler( LOG_DIR / "oplab_logger.log", maxBytes=5_000_000, backupCount=5, encoding="utf-8" ) arquivo.setFormatter(fmt) root.addHandler(arquivo) # ============================================================================ # CONFIGURAÇÃO # ============================================================================ @dataclass class Config: oplab_base_url: str = "https://api.oplab.com.br/v3" oplab_auth_url: str = "https://api.oplab.com.br/v3/domain/users/authenticate" oplab_token: str = os.getenv("OPLAB_TOKEN", "") oplab_email: str = os.getenv("OPLAB_EMAIL", "jrussi@gmail.com") oplab_password: str = os.getenv("OPLAB_PASSWORD", "jkiller77") underlying: str = os.getenv("OPLAB_UNDERLYING", "BOVA11") irate: float = float(os.getenv("OPLAB_IRATE", "14")) intervalo_checagem_seg: int = int(os.getenv("OPLAB_INTERVALO_SEG", "300")) # 5 min horario_inicio: dtime = dtime(10, 15) horario_fim: dtime = dtime(16, 45) todos_vencimentos: bool = os.getenv("OPLAB_TODOS_VENCIMENTOS", "false").lower() == "true" # Se preenchido, filtra só séries com days_to_maturity < esse valor # (tem prioridade sobre todos_vencimentos). Deixe vazio (env var) para não filtrar. _max_dias_raw: str = os.getenv("OPLAB_MAX_DIAS_VENCIMENTO", "24") max_dias_vencimento: Optional[int] = int(_max_dias_raw) if _max_dias_raw.strip() else None log_csv_dir: str = str(DATA_DIR) log_csv_prefix: str = "oplab_options_log" # ============================================================================ # CLIENTE OPLAB # ============================================================================ class OpLabClient: """ Login -> token -> header Access-Token em cada chamada, com reautenticação automática em caso de 401 (fluxo confirmado no OpLab.pas). """ def __init__(self, cfg: Config): self.cfg = cfg self.session = requests.Session() self.session.headers.update({"Accept": "application/json"}) self.token: Optional[str] = cfg.oplab_token or None if self.token: self.session.headers["Access-Token"] = self.token logging.info("Usando token fixo fornecido (OPLAB_TOKEN).") else: self.authenticate() def authenticate(self): if not (self.cfg.oplab_email and self.cfg.oplab_password): raise RuntimeError( "Defina OPLAB_TOKEN, ou OPLAB_EMAIL + OPLAB_PASSWORD (no .env) para fazer login." ) resp = requests.post( self.cfg.oplab_auth_url, json={"email": self.cfg.oplab_email, "password": self.cfg.oplab_password}, timeout=15, ) resp.raise_for_status() data = resp.json() token = data.get("access-token") or data.get("access_token") or data.get("token") if not token: raise RuntimeError(f"Login OK mas não achei o token na resposta: {data}") self.token = token self.session.headers["Access-Token"] = self.token logging.info("Autenticado na OpLab com sucesso (login por e-mail/senha).") def _get(self, path: str, params: Optional[dict] = None, _retry: bool = True) -> dict: url = f"{self.cfg.oplab_base_url}{path}" resp = self.session.get(url, params=params, timeout=20) if resp.status_code == 401 and _retry and self.cfg.oplab_email and self.cfg.oplab_password: logging.warning("Token expirado/inválido (401). Reautenticando e tentando de novo...") self.authenticate() return self._get(path, params=params, _retry=False) resp.raise_for_status() return resp.json() def get_series(self, symbol: str) -> dict: params = {"bs": "true", "irate": self.cfg.irate} return self._get(f"/market/instruments/series/{symbol}", params=params) # ============================================================================ # PARSING # ============================================================================ def _primeiro_valor(*fontes_e_chaves) -> Optional[float]: for fonte, chaves in fontes_e_chaves: if not fonte: continue for chave in chaves: if fonte.get(chave) is not None: return fonte.get(chave) return None def extrair_spot(data: dict) -> dict: return { "symbol": data.get("symbol"), "close": data.get("close"), "bid": data.get("bid"), "ask": data.get("ask"), } def _normalizar_opcao(opt: dict, tipo: str, strike: float, due_date: str, dtm) -> Optional[dict]: if not opt: return None greeks = opt.get("greeks") or opt.get("bs") or {} delta = _primeiro_valor((opt, ["delta"]), (greeks, ["delta"])) gamma = _primeiro_valor((opt, ["gamma"]), (greeks, ["gamma"])) theta = _primeiro_valor((opt, ["theta"]), (greeks, ["theta"])) vega = _primeiro_valor((opt, ["vega"]), (greeks, ["vega"])) iv = _primeiro_valor( (opt, ["iv", "implied_volatility", "volatility"]), (greeks, ["iv", "implied_volatility", "volatility"]), ) return { "symbol": opt.get("symbol"), "tipo": tipo, "strike": strike, "due_date": due_date, "days_to_maturity": dtm, "bid": opt.get("bid"), "ask": opt.get("ask"), "close": opt.get("close"), "delta": delta, "gamma": gamma, "theta": theta, "vega": vega, "iv": iv, } def iter_opcoes(data: dict): for serie in data.get("series") or []: due_date = serie.get("due_date") dtm = serie.get("days_to_maturity") for strike_obj in serie.get("strikes") or []: strike = strike_obj.get("strike") call = _normalizar_opcao(strike_obj.get("call"), "CALL", strike, due_date, dtm) if call: yield call put = _normalizar_opcao(strike_obj.get("put"), "PUT", strike, due_date, dtm) if put: yield put # ============================================================================ # LOGGER # ============================================================================ class OptionsLogger: def __init__(self, cfg: Config): self.cfg = cfg self.client = OpLabClient(cfg) self._file = None self._writer = None self._data_arquivo_atual = None # date do arquivo aberto no momento self._abrir_arquivo_do_dia_se_necessario() def _caminho_csv(self, dia) -> Path: return Path(self.cfg.log_csv_dir) / f"{self.cfg.log_csv_prefix}_{dia:%Y-%m-%d}.csv" def _abrir_arquivo_do_dia_se_necessario(self): """Abre (ou troca para) o CSV do dia atual. Cria 1 arquivo por dia.""" hoje = datetime.now().date() if hoje == self._data_arquivo_atual and self._file is not None: return # já está com o arquivo certo aberto if self._file is not None: self._file.close() logging.info(f"Virada de dia detectada. Fechando arquivo anterior.") caminho = self._caminho_csv(hoje) novo = not caminho.exists() self._file = open(caminho, "a", newline="", encoding="utf-8") self._writer = csv.writer(self._file) if novo: self._writer.writerow([ "timestamp", "spot", "symbol", "tipo", "strike", "due_date", "days_to_maturity", "bid", "ask", "close", "delta", "gamma", "theta", "vega", "iv", ]) self._data_arquivo_atual = hoje logging.info(f"Gravando no arquivo do dia: {caminho}") def coletar_uma_vez(self): self._abrir_arquivo_do_dia_se_necessario() ts = datetime.now().isoformat(timespec="seconds") try: data = self.client.get_series(self.cfg.underlying) except Exception as e: logging.error(f"Falha ao buscar série de opções: {e}") return spot_info = extrair_spot(data) spot = spot_info.get("close") registros = list(iter_opcoes(data)) if self.cfg.max_dias_vencimento is not None: registros = [ r for r in registros if r["days_to_maturity"] is not None and r["days_to_maturity"] < self.cfg.max_dias_vencimento ] elif not self.cfg.todos_vencimentos and registros: venc_mais_proximo = sorted(set( r["due_date"] for r in registros if r["due_date"] ))[0] registros = [r for r in registros if r["due_date"] == venc_mais_proximo] for r in registros: self._writer.writerow([ ts, spot, r["symbol"], r["tipo"], r["strike"], r["due_date"], r["days_to_maturity"], r["bid"], r["ask"], r["close"], r["delta"], r["gamma"], r["theta"], r["vega"], r["iv"], ]) self._file.flush() caminho_atual = self._caminho_csv(self._data_arquivo_atual) logging.info(f"{ts} | spot={spot} | {len(registros)} opções gravadas em {caminho_atual}") def fechar(self): if self._file is not None: self._file.close() # ============================================================================ # EXECUÇÃO # ============================================================================ def valida_credenciais(cfg: Config): if not cfg.oplab_token and not (cfg.oplab_email and cfg.oplab_password): logging.error( "Defina OPLAB_TOKEN, ou OPLAB_EMAIL + OPLAB_PASSWORD, no .env ou nas " "variáveis de ambiente antes de rodar." ) sys.exit(1) def run_once(cfg: Config, force: bool = False): """ Faz uma única coleta e sai. É isso que o systemd chama a cada 5 min. Checa a janela de horário aqui também (segurança extra, caso o timer dispare fora do intervalo por algum motivo) e em dia de fim de semana. Use force=True (--force) para pular essas checagens e testar manualmente fora do horário/dia de pregão (ex.: fim de semana). O mercado pode estar fechado, então os preços/gregas retornados serão os do último fechamento, não em tempo real - normal para esse teste. """ agora = datetime.now() if not force: if agora.weekday() >= 5: # 5=sábado, 6=domingo logging.info("Hoje é fim de semana. Nada a coletar. (use --once --force para testar mesmo assim)") return if not (cfg.horario_inicio <= agora.time() <= cfg.horario_fim): logging.info( f"Fora da janela de coleta ({cfg.horario_inicio}-{cfg.horario_fim}). " f"Agora: {agora.time()}. Ignorando esta chamada. " f"(use --once --force para testar mesmo assim)" ) return logger = OptionsLogger(cfg) try: logger.coletar_uma_vez() finally: logger.fechar() def run_loop(cfg: Config): """Loop contínuo, respeitando a janela de horário. Usado no modo manual/teste.""" logger = OptionsLogger(cfg) logging.info( f"Loop iniciado. Ativo: {cfg.underlying}. Janela: {cfg.horario_inicio} - " f"{cfg.horario_fim}, intervalo: {cfg.intervalo_checagem_seg}s. " f"Salvando em {cfg.log_csv_dir}/{cfg.log_csv_prefix}_AAAA-MM-DD.csv" ) try: while True: agora = datetime.now() if agora.time() > cfg.horario_fim: logging.info("Fora da janela de horário (após o fim). Encerrando.") break if agora.time() < cfg.horario_inicio: logging.info("Ainda não chegou o horário de início. Aguardando...") time.sleep(min(cfg.intervalo_checagem_seg, 30)) continue logger.coletar_uma_vez() time.sleep(cfg.intervalo_checagem_seg) except KeyboardInterrupt: logging.info("Interrompido manualmente.") finally: logger.fechar() def debug_inspect(cfg: Config): import json client = OpLabClient(cfg) data = client.get_series(cfg.underlying) campos_raiz = {k: v for k, v in data.items() if k != "series"} print("--- Campos na raiz (dados do ativo-objeto) ---") print(json.dumps(campos_raiz, indent=2, ensure_ascii=False)) series = data.get("series") or [] if series: strikes = series[0].get("strikes") or [] if strikes: print("\n--- Exemplo de 1 strike completo (call + put) ---") print(json.dumps(strikes[0], indent=2, ensure_ascii=False)) # ============================================================================ # ENTRYPOINT # ============================================================================ if __name__ == "__main__": configurar_logging() parser = argparse.ArgumentParser(description="Logger de spot + opções (OpLab) - BOVA11.") grupo = parser.add_mutually_exclusive_group(required=True) grupo.add_argument("--inspect", action="store_true", help="Inspeciona os campos crus da API e sai.") grupo.add_argument("--once", action="store_true", help="Faz UMA coleta e sai (usado pelo systemd).") grupo.add_argument("--run", action="store_true", help="Roda o loop de coleta (uso manual/teste).") parser.add_argument("--force", action="store_true", help="Com --once: ignora checagem de horário/fim de semana (uso: testes manuais).") args = parser.parse_args() cfg = Config() valida_credenciais(cfg) if args.inspect: debug_inspect(cfg) elif args.once: run_once(cfg, force=args.force) elif args.run: run_loop(cfg)