Spaces:
Sleeping
Sleeping
File size: 5,122 Bytes
55d9355 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """Use the imported model snapshot; local names alone do not establish identity."""
import hashlib
import json
from pathlib import Path
import numpy as np
import pandas as pd
from . import get_model, list_available_models
from .model_data import ModelData
from .anos import detectar_ano_base, periodo_variavel_ano
def context_key(row):
terrain = row.get('evolutivo', {}).get('terreno', {})
context = terrain.get('contextoModelo') or {}
return row.get('modelo', ''), hashlib.sha256(json.dumps(context, sort_keys=True, ensure_ascii=False).encode()).hexdigest()[:12]
def matching_local(row):
name = row.get('modelo')
context = row.get('evolutivo', {}).get('terreno', {}).get('contextoModelo') or {}
if name not in list_available_models():
return None, 'Modelo indisponível no catálogo.'
local = get_model(name)
if not context:
return None, 'O arquivo não preserva o contexto necessário para confirmar a versão do modelo do catálogo.'
prediction = context.get('prediction') or {}
names, params = prediction.get('names') or [], prediction.get('params') or []
sm = local.modelos_sm if local else None
try:
local_names = list(map(str, sm.model.exog_names))
same = names == local_names and len(params) == len(sm.params) and np.allclose(params, sm.params, rtol=1e-12, atol=1e-12)
# Equal parameters are necessary, but not sufficient: verify fitted data too.
x = context.get('transformacoes', {}).get('X')
y = context.get('transformacoes', {}).get('y')
portable_x = pd.DataFrame(x).reindex(columns=names, fill_value=1).to_numpy(dtype=float)
portable_y = np.asarray(y, dtype=float).reshape(-1)
same = same and portable_x.shape == sm.model.exog.shape and portable_y.shape == sm.model.endog.shape
same = same and np.allclose(portable_x, sm.model.exog, rtol=1e-12, atol=1e-12) and np.allclose(portable_y, sm.model.endog, rtol=1e-12, atol=1e-12)
covariance = np.asarray(prediction.get('covariance'),dtype=float)
same = same and covariance.shape == np.asarray(sm.cov_params()).shape and np.allclose(covariance,sm.cov_params(),rtol=1e-12,atol=1e-12)
same = same and context.get('transformacoes',{}).get('info') == local.formatted_top_transformation_info
if same:
return local, ''
except (AttributeError, TypeError, ValueError, KeyError):
pass
return None, 'A correspondência da versão do catálogo não foi confirmada com os coeficientes e a amostra salvos; os anexos desse modelo não serão usados.'
def snapshot_model(row):
context = row.get('evolutivo', {}).get('terreno', {}).get('contextoModelo') or {}
prediction = context.get('prediction') or {}
names, params = prediction.get('names') or [], prediction.get('params') or []
if not names or len(names) != len(params):
return None
raw = context.get('dados', {}).get('df') or {}
try:
df = pd.DataFrame(**raw) if isinstance(raw, dict) and 'columns' in raw else pd.DataFrame(raw)
except (ValueError, TypeError):
df = pd.DataFrame()
transforms = context.get('transformacoes') or {}
info = transforms.get('info') or []
mapping = dict(item.split(': ', 1) for item in info if isinstance(item, str) and ': ' in item)
# Transcribe the fitted expression; do not fit or evaluate a new model.
terms = []
for name, param in zip(names, params):
coefficient = format(float(param), '.12g')
if name == 'const':
terms.append(coefficient)
else:
transformed = mapping.get(name, '(x)').replace('x', f'[{name}]')
terms.append(f'({coefficient} × {transformed})')
dependent = next((item.split(': ', 1) for item in info if isinstance(item, str) and ': ' in item and item.split(': ', 1)[0] not in names), ['Y', 'y'])
lhs = dependent[1].replace('y', dependent[0])
evidence = context.get('normativeEvidence') or {}
annual_names = [str(name) for name in names if periodo_variavel_ano(str(name))]
reference = context.get('referencia_temporal')
base_year = detectar_ano_base(df, annual_names, reference)
if base_year is None and reference is None:
# Legacy EVO contexts omitted the declaration. A name alone is not
# enough to borrow it: matching_local verifies the fitted model/data.
local, _ = matching_local(row)
if local is not None:
base_year = local.ano_base
return ModelData(
nome=row.get('modelo') or context.get('name') or 'Modelo importado', path=Path('contexto_importado'),
xy_preview=df, tabelas_coef=pd.DataFrame({'coef': params}, index=names),
modelos_resumos={'n': evidence.get('n', len(df)), 'k': evidence.get('k', len(names)-1),
'r2': evidence.get('r2'), 'equacao': f'{lhs} = ' + ' + '.join(terms)},
formatted_top_transformation_info=info,
variaveis_dicotomicas=transforms.get('dicotomicas') or [],
periodo_dados_mercado=context.get('periodo_dados_mercado') or {},
ano_base=base_year,
)
|