amarorn / docs /datalake-e-pipelines.md
beAnalytic's picture
feat: sync main with feature/superbet-live-inplay
16c19b8 verified
|
Raw
History Blame Contribute Delete
7.53 kB
# Datalake e pipelines
## Camadas do datalake
```mermaid
flowchart LR
RSS[RSS feeds] --> Bronze
Bronze --> Silver
Silver --> Gold
Fixtures[Fixtures CSV/TXT] --> Gold
Gold --> JSONL[JSONL treino LM]
```
| Camada | Path | Conteúdo |
|--------|------|----------|
| **Bronze** | `data/lake/bronze/` | Feed RSS bruto, metadados, parquet por fonte/data |
| **Silver** | `data/lake/silver/` | Artigos limpos, times, sentimento, menções lesão |
| **Gold** | `data/lake/gold/` | Contexto agregado por confronto + labels |
| **Fixtures** | `data/lake/fixtures/` | `world_cup_YYYY.parquet`, brasileirao |
---
## Fontes RSS
Configuração: [`data/sources.yaml`](../data/sources.yaml)
```yaml
sources:
- id: globo_esporte
name: Globo Esporte
url: https://...
```
| ID | Portal |
|----|--------|
| `globo_esporte` | Globo Esporte |
| `espn_br` | ESPN Brasil |
| `uol_esporte` | UOL Esporte |
| `fogaonet` | Fogaonet |
| `gazeta_esportiva` | Gazeta Esportiva |
```bash
collect-news --list-sources
Por padrão, até **~1000 artigos únicos** por sync (14 fontes RSS + feeds Google Notícias; limite `RSS_MAX_ENTRIES_PER_SOURCE=160`).
```
---
## Comandos CLI
### Ingestão
| Comando | Descrição |
|---------|-----------|
| `collect-news` | RSS → bronze |
| `daily-sync` | Coleta + pipeline silver |
| `import-brasileirao` | Fixtures Brasileirão (openfootball) |
| `import-world-cup` | Fixtures Copa 1930–2022 |
**Copa do Mundo:**
```bash
import-world-cup --list
import-world-cup --missing-only
import-world-cup --seasons 1970 1982 2022
import-world-cup --force
```
### Transformação
| Comando | Descrição |
|---------|-----------|
| `run-pipeline silver` | bronze → silver |
| `run-pipeline gold --season 2024` | silver + fixtures → gold |
| `run-pipeline export` | exporta JSONL para treino LM |
### Previsão
| Comando | Descrição |
|---------|-----------|
| `predict-round` | Rodada `data/rounds/current.json` |
| `predict-wc` | Palpites WC (`--json`, `--home`, `--away`) |
| `study-wc-model` | Relatório JSON do ensemble |
### Odds e valor
| Comando | Descrição |
|---------|-----------|
| `fetch-wc-odds` | Odds ao vivo → `wc_2026_odds.json` |
| `value-wc-odds` | EV positivo vs modelo |
### Benchmark
```bash
benchmark-wc-models --eval-season 2022
benchmark-wc-models --eval-season 2022 --mlflow
```
---
## Arquivos de rodada
### `data/rounds/wc_2026.json`
```json
{
"season": 2026,
"competition": "Copa do Mundo",
"phase": "group",
"round": 1,
"matches": [
{ "home_team": "Brasil", "away_team": "Marrocos", "phase": "group", "group": "G" }
]
}
```
### `data/rounds/current.json`
Rodada ativa do Brasileirão para `predict-round` e `GET /round/predict`.
---
## Pipeline gold (contexto por jogo)
[`pipelines/gold.py`](../pipelines/gold.py) agrega:
- Artigos silver dos times
- Posição, forma, H2H (fixtures Brasileirão)
- Features para baseline e futura LM
Modo live (`live_mode=True` na API): usa notícias recentes sem label.
---
## Validação histórica WC
[`pipelines/wc_validate.py`](../pipelines/wc_validate.py)
- Lista edições e jogos
- `validate_historical_match`: `before_date = match_date`, filtro estrito `<`
- Sanitiza `NaN` em `group_name` (mata-mata)
---
## Agendamento
- Cron local: [`scripts/cron_collect.sh`](../scripts/cron_collect.sh)
- GitHub Actions: [`.github/workflows/daily-collect.yml`](../.github/workflows/daily-collect.yml)
---
## Export para treino LM
```python
from models.dataset import export_jsonl
export_jsonl("data/training/bolao_train.jsonl")
```
Formato JSONL: contexto textual + label `1`/`X`/`2`.
Treino:
```bash
pip install -e ".[ml]"
python -m models.train
```
Integração recomendada: [Unsloth](https://github.com/unslothai/unsloth).
---
## Storage: Parquet local + DuckDB + BigQuery
### Camada local (padrão)
| Dado | Path | Formato |
|------|------|---------|
| Notícias | `data/lake/bronze`, `silver`, `gold` | Parquet particionado |
| Fixtures | `data/lake/fixtures/` | Parquet |
| Stats Sofascore | `data/lake/sofascore/match_stats.parquet` | Parquet único |
| Auditoria Sofascore | `data/lake/sofascore/*_stats.json` | JSON por jogo (opcional) |
Cada ingestão Sofascore grava JSON + upsert no parquet. Para reconstruir o parquet a partir dos JSONs:
```bash
ingest-sofascore --compact-parquet
ingest-sofascore --compact-parquet --json
```
### Consultas SQL locais (DuckDB)
```bash
pip install -e ".[analytics]"
lake-query --preset summary
lake-query --preset team-xg --team Brasil --limit 10
lake-query --sql "SELECT home_team, AVG(home_xg) AS avg_xg FROM sofascore GROUP BY 1 ORDER BY 2 DESC LIMIT 5"
```
Views disponíveis: `bronze`, `silver`, `gold`, `fixtures` / `silver_fixtures`, `sofascore` / `silver_sofascore`.
### Quando ativar BigQuery + GCS
| Gatilho | Ação |
|---------|------|
| Lake > ~500 MB ou histórico longo de notícias | `sync-gcp --layer all` |
| Múltiplos ambientes (dev/staging/prod) | Sync periódico para GCS |
| Dashboards / SQL ad hoc na nuvem | BigQuery como camada analítica |
| Sofascore no BQ para joins com fixtures | `sync-gcp --layer silver_sofascore` |
```bash
pip install -e ".[gcp]"
# .env: GCP_PROJECT, BQ_DATASET, GCS_BUCKET (opcional)
sync-gcp --list-layers
sync-gcp --layer all --truncate
sync-gcp --layer silver_sofascore --truncate
sync-gcp --layer bronze_sofascore --truncate
sync-gcp --layer gold_wc --truncate
```
Tabelas BigQuery (medalhão em `sports_news_lake`):
| Camada | Tabela BQ |
|--------|-----------|
| Bronze notícias | `bronze_articles` |
| Bronze Sofascore | `bronze_sofascore_events` |
| Silver notícias | `silver_articles` |
| Silver Sofascore | `silver_sofascore_match_stats` |
| Silver fixtures WC | `silver_fixtures_results` |
| Gold bolão | `gold_bolao_context` |
| Gold WC features | `gold_wc_match_features` |
Tabelas legadas (`sofascore_match_stats`, `fixtures_results`) podem ser removidas manualmente no console BQ após migração.
**Não substitua** o Parquet local — a API e os modelos ML continuam lendo arquivos no volume (`LAKE_ROOT`). O BigQuery é camada analítica/backup, não storage primário de runtime.
### GCS (camada de objetos — padrão GCP)
No desenho correto, o fluxo é **Parquet local → GCS → BigQuery**:
```mermaid
flowchart LR
Local[data/lake Parquet] --> GCS[gs://bucket/lake/...]
GCS --> BQ[BigQuery sports_news_lake]
```
Layout medalhão no bucket (quando `GCS_BUCKET` está definido):
```
gs://{GCS_BUCKET}/
└── lake/
├── bronze/articles/...
├── bronze/sofascore/events.parquet
├── silver/articles/...
├── silver/sofascore/match_stats.parquet
├── silver/fixtures/world_cup_fixtures.parquet
├── gold/bolao/...
└── gold/wc/match_features.parquet
```
**Configuração no GCP** (service account `modelo-cp@beanalytic-dev.iam.gserviceaccount.com`):
1. Criar bucket, ex.: `beanalytic-dev-sports-news-lake` (região `US`, mesma do BQ)
2. IAM no bucket: `Storage Object Admin` (ou `Storage Admin`)
3. `.env`: `GCS_BUCKET=beanalytic-dev-sports-news-lake`
4. `sync-gcp --layer all --truncate` — sobe Parquet ao GCS e carrega no BQ a partir de `gs://`
Sem `GCS_BUCKET`, o `sync-gcp` usa **fallback direto** (Parquet local → BQ), útil em dev quando a SA ainda não tem permissão no Storage.
## Escalabilidade GCP (opcional)
```bash
pip install -e ".[gcp]"
```
Variáveis: `GCP_PROJECT`, `BQ_DATASET`, `GCS_BUCKET` — ver `.env.example`.