340 lines
15 KiB
Python
340 lines
15 KiB
Python
"""Motor de regras de auditoria do Dashboard Contábil — cada `regra_*` é uma
|
|
função pura (`ResultadoExtracao` da apuração atual + histórico já persistido
|
|
da mesma empresa → `list[AchadoDetectado]`), sem tocar no ORM. Cobrem só o
|
|
que é derivável do próprio balancete/DRE anexado (ver CLAUDE.md do pacote) —
|
|
qualquer checagem que dependa de sistemas externos (Questor, extratos,
|
|
folha) fica fora de propósito, por decisão do usuário (ferramenta analítica,
|
|
não operacional)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from .modelos import AchadoDetectado, LinhaBalanceteExtraida, ResultadoExtracao, SnapshotHistorico
|
|
from .parser import LINHA_DRE_CUSTOS_TOTAIS, LINHA_DRE_RECEITA_LIQUIDA
|
|
|
|
SEVERIDADE_ALTA = "alta"
|
|
SEVERIDADE_MEDIA = "media"
|
|
SEVERIDADE_BAIXA = "baixa"
|
|
|
|
TOLERANCIA_CENTAVOS = Decimal("0.05")
|
|
|
|
# Trechos de descrição que, segundo o roteiro de conferência (ITD-FISCO-7513),
|
|
# devem ficar zeradas todo mês — ex.: "o saldo de adiantamento de salários e
|
|
# empréstimos a funcionários deve estar zerado todos os meses". Deliberadamente
|
|
# restrito a contas cujo saldo residual é sempre um sinal de pendência (não
|
|
# inclui, por exemplo, "ADIANTAMENTOS DE FÉRIAS"/"13º SALÁRIO", que legitimamente
|
|
# carregam saldo entre um mês e o pagamento seguinte).
|
|
TRECHOS_CONTA_DEVERIA_ZERAR = ["ADIANTAMENTOS DE SALÁRIOS", "ADIANTAMENTO DE SALÁRIOS"]
|
|
|
|
VARIACAO_LIMIAR_PERCENTUAL = Decimal("0.5") # 50%
|
|
VARIACAO_VALOR_MINIMO = Decimal("1000") # ignora variações abaixo disso, mesmo que %-mente grandes
|
|
|
|
|
|
def _mes_consecutivo(competencia_atual: date, competencia_anterior: date) -> bool:
|
|
return (competencia_atual.year * 12 + competencia_atual.month) - (
|
|
competencia_anterior.year * 12 + competencia_anterior.month
|
|
) == 1
|
|
|
|
|
|
def _isola_mes_dre(valor_ytd_atual: Decimal, valor_ytd_anterior: Decimal | None, competencia_atual: date) -> Decimal:
|
|
"""A DRE do relatório é acumulada desde janeiro — isola o mês corrente
|
|
subtraindo o acumulado do mês anterior, exceto em janeiro (onde o
|
|
acumulado já é o próprio mês)."""
|
|
if competencia_atual.month == 1 or valor_ytd_anterior is None:
|
|
return valor_ytd_atual
|
|
return valor_ytd_atual - valor_ytd_anterior
|
|
|
|
|
|
def regra_balanceamento_ativo_passivo(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
ativo = next((c for c in atual.contas if c.codigo == "1"), None)
|
|
passivo = next((c for c in atual.contas if c.codigo == "2"), None)
|
|
if ativo is None or passivo is None:
|
|
return []
|
|
diferenca = ativo.saldo_atual + passivo.saldo_atual # passivo já vem negativo no relatório
|
|
if abs(diferenca) <= TOLERANCIA_CENTAVOS:
|
|
return []
|
|
return [
|
|
AchadoDetectado(
|
|
regra="balanceamento_ativo_passivo",
|
|
severidade=SEVERIDADE_ALTA,
|
|
titulo="Ativo não bate com Passivo",
|
|
mensagem=(
|
|
f"Saldo do Ativo (R$ {ativo.saldo_atual}) não coincide com o do Passivo "
|
|
f"(R$ {-passivo.saldo_atual}) — diferença de R$ {abs(diferenca)}."
|
|
),
|
|
valor_referencia=diferenca,
|
|
)
|
|
]
|
|
|
|
|
|
def regra_debito_credito_divergente(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
raizes = [c for c in atual.contas if "." not in c.codigo]
|
|
soma_debito = sum((c.debito for c in raizes), Decimal(0))
|
|
soma_credito = sum((c.credito for c in raizes), Decimal(0))
|
|
diferenca = soma_debito - soma_credito
|
|
if abs(diferenca) <= TOLERANCIA_CENTAVOS:
|
|
return []
|
|
return [
|
|
AchadoDetectado(
|
|
regra="debito_credito_divergente",
|
|
severidade=SEVERIDADE_ALTA,
|
|
titulo="Débito total diferente do crédito total",
|
|
mensagem=(
|
|
f"Soma de débitos do período (R$ {soma_debito}) não bate com a soma de "
|
|
f"créditos (R$ {soma_credito}) — diferença de R$ {abs(diferenca)}."
|
|
),
|
|
valor_referencia=diferenca,
|
|
)
|
|
]
|
|
|
|
|
|
def regra_saldo_negativo_caixa(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
achados = []
|
|
for conta in atual.contas:
|
|
if conta.codigo.startswith("1.01.01.001") and conta.saldo_atual < 0:
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="saldo_negativo_caixa",
|
|
severidade=SEVERIDADE_ALTA,
|
|
titulo="Saldo de caixa negativo",
|
|
mensagem=f'Conta "{conta.descricao}" ({conta.codigo}) com saldo negativo de R$ {conta.saldo_atual}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_conta_transitoria_com_saldo(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
achados = []
|
|
for conta in atual.contas:
|
|
descricao_normalizada = conta.descricao.upper()
|
|
if ("TRANSIT" in descricao_normalizada) and conta.saldo_atual != 0:
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="conta_transitoria_com_saldo",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Conta transitória com saldo",
|
|
mensagem=f'Conta transitória "{conta.descricao}" ({conta.codigo}) deveria estar zerada e está com R$ {conta.saldo_atual}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_conta_deveria_zerar(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
achados = []
|
|
for conta in atual.contas:
|
|
descricao_normalizada = conta.descricao.upper()
|
|
if conta.saldo_atual == 0:
|
|
continue
|
|
if any(trecho in descricao_normalizada for trecho in TRECHOS_CONTA_DEVERIA_ZERAR):
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="conta_deveria_zerar",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Conta que deveria estar zerada",
|
|
mensagem=f'Conta "{conta.descricao}" ({conta.codigo}) normalmente fica zerada todo mês e está com saldo de R$ {conta.saldo_atual}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def _eh_conta_redutora(conta: LinhaBalanceteExtraida) -> bool:
|
|
return conta.descricao.strip().startswith("(-)")
|
|
|
|
|
|
def regra_saldo_sinal_invertido(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
achados = []
|
|
for conta in atual.contas:
|
|
if conta.tipo != "A" or conta.saldo_atual == 0 or _eh_conta_redutora(conta):
|
|
continue
|
|
if conta.codigo.startswith("1.") and conta.saldo_atual < 0:
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="saldo_sinal_invertido",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Conta do Ativo com saldo credor",
|
|
mensagem=f'Conta do Ativo "{conta.descricao}" ({conta.codigo}) está com saldo credor de R$ {-conta.saldo_atual}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
elif conta.codigo.startswith("2.") and conta.saldo_atual > 0:
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="saldo_sinal_invertido",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Conta do Passivo com saldo devedor",
|
|
mensagem=f'Conta do Passivo "{conta.descricao}" ({conta.codigo}) está com saldo devedor de R$ {conta.saldo_atual}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_descricao_generica(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
achados = []
|
|
for conta in atual.contas:
|
|
if conta.descricao.strip().upper() == "DIVERSOS" and conta.saldo_atual != 0:
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="descricao_generica",
|
|
severidade=SEVERIDADE_BAIXA,
|
|
titulo='Conta com descrição genérica ("DIVERSOS")',
|
|
mensagem=f'Conta {conta.codigo} está descrita apenas como "DIVERSOS", com saldo de R$ {conta.saldo_atual} — recomenda-se detalhar em conta analítica própria.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_variacao_atipica_saldo(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
if not historico:
|
|
return []
|
|
anterior = historico[0]
|
|
achados = []
|
|
for conta in atual.contas:
|
|
if conta.tipo != "A":
|
|
continue
|
|
saldo_anterior = anterior.contas.get(conta.codigo)
|
|
if saldo_anterior is None or saldo_anterior == 0:
|
|
continue
|
|
diferenca = conta.saldo_atual - saldo_anterior
|
|
if abs(diferenca) < VARIACAO_VALOR_MINIMO:
|
|
continue
|
|
variacao_percentual = abs(diferenca) / abs(saldo_anterior)
|
|
if variacao_percentual < VARIACAO_LIMIAR_PERCENTUAL:
|
|
continue
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="variacao_atipica_saldo",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Variação atípica de saldo",
|
|
mensagem=(
|
|
f'Conta "{conta.descricao}" ({conta.codigo}) variou {variacao_percentual:.0%} em relação a '
|
|
f"{anterior.competencia:%m/%Y} (de R$ {saldo_anterior} para R$ {conta.saldo_atual})."
|
|
),
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=diferenca,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_variacao_atipica_dre(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
if len(historico) < 2:
|
|
return []
|
|
anterior, retrasado = historico[0], historico[1]
|
|
if not _mes_consecutivo(anterior.competencia, retrasado.competencia):
|
|
return []
|
|
achados = []
|
|
for linha in atual.linhas_dre:
|
|
valor_anterior_ytd = anterior.linhas_dre.get(linha.descricao)
|
|
valor_retrasado_ytd = retrasado.linhas_dre.get(linha.descricao)
|
|
if valor_anterior_ytd is None:
|
|
continue
|
|
mes_atual = _isola_mes_dre(linha.valor, valor_anterior_ytd, atual.cabecalho.periodo_fim)
|
|
mes_anterior = _isola_mes_dre(valor_anterior_ytd, valor_retrasado_ytd, anterior.competencia)
|
|
if mes_anterior == 0:
|
|
continue
|
|
diferenca = mes_atual - mes_anterior
|
|
if abs(diferenca) < VARIACAO_VALOR_MINIMO:
|
|
continue
|
|
variacao_percentual = abs(diferenca) / abs(mes_anterior)
|
|
if variacao_percentual < VARIACAO_LIMIAR_PERCENTUAL:
|
|
continue
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="variacao_atipica_dre",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Variação atípica na DRE",
|
|
mensagem=(
|
|
f'Linha "{linha.descricao}" da DRE variou {variacao_percentual:.0%} em relação ao mês anterior '
|
|
f"(de R$ {mes_anterior} para R$ {mes_atual})."
|
|
),
|
|
valor_referencia=diferenca,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_percentual_custo_receita_atipico(
|
|
atual: ResultadoExtracao, historico: list[SnapshotHistorico]
|
|
) -> list[AchadoDetectado]:
|
|
if len(historico) < 2:
|
|
return []
|
|
anterior, retrasado = historico[0], historico[1]
|
|
if not _mes_consecutivo(anterior.competencia, retrasado.competencia):
|
|
return []
|
|
|
|
def _linha(resultado_linhas: dict[str, Decimal], nome: str) -> Decimal | None:
|
|
return resultado_linhas.get(nome)
|
|
|
|
linhas_atual = {l.descricao: l.valor for l in atual.linhas_dre}
|
|
receita_atual_ytd = _linha(linhas_atual, LINHA_DRE_RECEITA_LIQUIDA)
|
|
custos_atual_ytd = _linha(linhas_atual, LINHA_DRE_CUSTOS_TOTAIS)
|
|
receita_anterior_ytd = _linha(anterior.linhas_dre, LINHA_DRE_RECEITA_LIQUIDA)
|
|
custos_anterior_ytd = _linha(anterior.linhas_dre, LINHA_DRE_CUSTOS_TOTAIS)
|
|
receita_retrasado_ytd = _linha(retrasado.linhas_dre, LINHA_DRE_RECEITA_LIQUIDA)
|
|
custos_retrasado_ytd = _linha(retrasado.linhas_dre, LINHA_DRE_CUSTOS_TOTAIS)
|
|
if None in (receita_atual_ytd, custos_atual_ytd, receita_anterior_ytd, custos_anterior_ytd):
|
|
return []
|
|
|
|
receita_mes_atual = _isola_mes_dre(receita_atual_ytd, receita_anterior_ytd, atual.cabecalho.periodo_fim)
|
|
custos_mes_atual = _isola_mes_dre(custos_atual_ytd, custos_anterior_ytd, atual.cabecalho.periodo_fim)
|
|
receita_mes_anterior = _isola_mes_dre(receita_anterior_ytd, receita_retrasado_ytd, anterior.competencia)
|
|
custos_mes_anterior = _isola_mes_dre(custos_anterior_ytd, custos_retrasado_ytd, anterior.competencia)
|
|
if receita_mes_atual == 0 or receita_mes_anterior == 0:
|
|
return []
|
|
|
|
pct_atual = abs(custos_mes_atual) / abs(receita_mes_atual)
|
|
pct_anterior = abs(custos_mes_anterior) / abs(receita_mes_anterior)
|
|
if pct_anterior == 0:
|
|
return []
|
|
variacao_percentual = abs(pct_atual - pct_anterior) / pct_anterior
|
|
if variacao_percentual < VARIACAO_LIMIAR_PERCENTUAL:
|
|
return []
|
|
return [
|
|
AchadoDetectado(
|
|
regra="percentual_custo_receita_atipico",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Percentual de custo sobre receita fora do padrão da empresa",
|
|
mensagem=(
|
|
f"O custo representou {pct_atual:.1%} da receita líquida do mês, contra {pct_anterior:.1%} no mês "
|
|
f"anterior — variação de {variacao_percentual:.0%} em relação ao padrão histórico desta empresa."
|
|
),
|
|
valor_referencia=pct_atual - pct_anterior,
|
|
)
|
|
]
|
|
|
|
|
|
REGRAS = [
|
|
regra_balanceamento_ativo_passivo,
|
|
regra_debito_credito_divergente,
|
|
regra_saldo_negativo_caixa,
|
|
regra_conta_transitoria_com_saldo,
|
|
regra_conta_deveria_zerar,
|
|
regra_saldo_sinal_invertido,
|
|
regra_descricao_generica,
|
|
regra_variacao_atipica_saldo,
|
|
regra_variacao_atipica_dre,
|
|
regra_percentual_custo_receita_atipico,
|
|
]
|
|
|
|
|
|
def gera_achados(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
"""`historico` deve vir ordenado da apuração mais recente pra mais antiga
|
|
(mesma empresa, competências anteriores à de `atual`)."""
|
|
achados: list[AchadoDetectado] = []
|
|
for regra in REGRAS:
|
|
achados.extend(regra(atual, historico))
|
|
return achados
|