movimientos_lib.py
Ruta original: docs/proyectos/consumo-interno-fifo/scripts/movimientos_lib.py
"""
Libreria reusable para el reporte de movimientos efectivos en un periodo, reconciliado
contra existencia a una fecha (Metodo C, validado 100% contra MPRO). Ver
docs/schema/inventario.md (raiz de este repo) para la clasificacion completa de tipos
de movimiento y el Metodo C -- este modulo es el codigo promovido de esa exploracion.
Usada por:
- 05_movimientos_periodo.py (CLI, genera CSV)
- streamlit_app_movimientos.py (reporte interactivo)
"""
import pandas as pd
import connection_205_trivasadb3 as c
PR_CVE_LEN = 10
TOLERANCIA = 0.5
# Traspaso entre almacenes / transferencia entre sucursales / en transito / traspaso a
# produccion -- confirmado que netean a 0 por producto (ver decision doc).
TIPOS_TRASPASO_TRANSFERENCIA = {
"100", "101", "500", "501", # transferencia entre sucursales
"102", "103", "502", "503", # entrada/salida de transito
"106", "107", "506", "507", # traspaso entre almacenes
"112", "113", "512", "513", # traspaso a produccion (entrada/salida, mismo neto)
}
# Anulaciones de los tipos de movimiento "reales" (no traspaso). Se excluyen por default.
# 201/903/981/983/053/905 agregados 2026-08-24 (round2): estaban fuera de este set y
# fuera de CLASIFICACION, asi que colaban como "SIN CLASIFICAR" en vez de excluirse. No
# rompian la reconciliacion (el neto de Movimiento ya las traia sumadas), pero ensuciaban
# el listado. Confirmado con reconciliar_y_ajustar: 0 productos sin cuadrar antes Y
# despues del cambio, para marzo 2026 y para el semestre completo (ene-jun 2026,
# 8,568 productos) -- ver docs/proyectos/consumo-interno-fifo/index.md.
TIPOS_ANULACION_REALES = {
"051", "061", "109", "115", "203", "509", "511", "515", "601", "701", "801", "973", "401",
"201", "903", "981", "983", "053", "905",
}
# Tm_Cve_Tipo_Movimiento -> (entrada_salida, categoria de negocio)
CLASIFICACION = {
"050": ("Entrada", "Compra"),
"052": ("Entrada", "Consignación"),
"108": ("Entrada", "Conversión"),
"114": ("Entrada", "Salida de Orden de Producción"),
"200": ("Entrada", "Rechazo"),
"202": ("Entrada", "Devolución de cliente"),
"902": ("Entrada", "Ajuste de inventario físico"),
"974": ("Entrada", "Merma (entrada)"),
"060": ("Salida", "Consumo interno"),
"508": ("Salida", "Conversión"),
"514": ("Salida", "Entrada a Orden de Producción"),
"510": ("Salida", "Merma"),
"600": ("Salida", "Venta (remisión)"),
"700": ("Salida", "Venta (nota de venta)"),
"800": ("Salida", "Refacciones / consumo"),
"904": ("Salida", "Ajuste de inventario físico"),
"972": ("Salida", "Reproceso"),
"980": ("Salida", "Ajuste MP bloquera"),
"982": ("Salida", "Ajuste MP viguera"),
"986": ("Salida", "Reposición física"),
"400": ("Salida", "Devolución a proveedor"),
# anulaciones (se incluyen en el listado solo si reconciliar_y_ajustar las reactiva)
"401": ("Entrada", "Anulación de Devolución a proveedor"),
"051": ("Salida", "Anulación de Compra"),
"061": ("Entrada", "Anulación de Consumo interno"),
"109": ("Salida", "Anulación de Conversión"),
"115": ("Salida", "Anulación de Salida de Orden de Producción"),
"203": ("Salida", "Anulación de Devolución de cliente"),
"509": ("Entrada", "Anulación de Conversión"),
"511": ("Entrada", "Anulación de Merma"),
"515": ("Entrada", "Anulación de Entrada a Orden de Producción"),
"601": ("Entrada", "Anulación de Venta (remisión)"),
"701": ("Entrada", "Anulación de Venta (nota de venta)"),
"801": ("Entrada", "Anulación de Refacciones / consumo"),
"973": ("Entrada", "Anulación de Reproceso"),
# agregados 2026-08-24 (round2) -- mismo criterio: invierten el signo del tipo que anulan
"201": ("Salida", "Anulación de Rechazo"), # anula 200 (Entrada)
"903": ("Salida", "Anulación de Ajuste de inventario físico"), # anula 902 (Entrada)
"981": ("Entrada", "Anulación de Ajuste MP bloquera"), # anula 980 (Salida)
"983": ("Entrada", "Anulación de Ajuste MP viguera"), # anula 982 (Salida)
"053": ("Salida", "Anulación de Consignación"), # anula 052 (Entrada)
"905": ("Entrada", "Anulación de Ajuste de inventario físico"), # anula 904 (Salida)
}
def existencia_a_fecha_metodo_c(fecha: str) -> pd.DataFrame:
"""Metodo validado 100% contra 3 exports reales de MPRO. fecha inclusive."""
df = c.q(f"""
SELECT Producto.Pr_Cve_Producto,
SUM(Movimiento.Mv_Cantidad_Control_1) AS Mv_Cantidad_Control_1,
SUM(Movimiento.Mv_Cantidad_Control_2) AS Mv_Cantidad_Control_2
FROM Movimiento
INNER JOIN Producto ON Movimiento.Pr_Cve_Producto = Producto.Pr_Cve_Producto
INNER JOIN Sucursal ON Movimiento.Sc_Cve_Sucursal = Sucursal.Sc_Cve_Sucursal
WHERE Sucursal.Em_Cve_Empresa = '0001' AND Movimiento.Mv_Fecha <= '{fecha}'
GROUP BY Producto.Pr_Cve_Producto
HAVING (SUM(Movimiento.Mv_Cantidad_Control_1) <> 0 OR SUM(Movimiento.Mv_Cantidad_Control_2) <> 0)
""")
df["Pr_Cve_Producto"] = df["Pr_Cve_Producto"].astype(str).str.strip().str.zfill(PR_CVE_LEN)
return df.groupby("Pr_Cve_Producto", as_index=False)["Mv_Cantidad_Control_1"].sum()
def verificar_traspasos_netean_cero(fecha_ini: str, fecha_fin: str) -> pd.DataFrame:
"""Chequeo de sanidad: productos donde traspaso/transferencia NO netea a 0 en el
periodo pedido (deberia venir vacio; si no, la exclusion de esas familias no es segura
para ese periodo especifico)."""
tipos = ",".join(f"'{t}'" for t in TIPOS_TRASPASO_TRANSFERENCIA)
return c.q(f"""
SELECT Pr_Cve_Producto, SUM(Mv_Cantidad_1) neto
FROM Movimiento
WHERE Mv_Fecha >= '{fecha_ini}' AND Mv_Fecha <= '{fecha_fin}'
AND Tm_Cve_Tipo_Movimiento IN ({tipos})
GROUP BY Pr_Cve_Producto
HAVING ABS(SUM(Mv_Cantidad_1)) > {TOLERANCIA}
""")
def extraer_movimientos(fecha_ini: str, fecha_fin: str, incluir_anulaciones=False) -> pd.DataFrame:
excluidos = set(TIPOS_TRASPASO_TRANSFERENCIA)
if not incluir_anulaciones:
excluidos |= TIPOS_ANULACION_REALES
excluidos_sql = ",".join(f"'{t}'" for t in excluidos)
filtro_cancelado = "" if incluir_anulaciones else "AND mv.Es_Cve_Estado <> 'CA'"
df = c.q(f"""
SELECT
mv.Mv_Folio, mv.Mv_ID, mv.Mv_Fecha,
mv.Pr_Cve_Producto, pr.Pr_Descripcion,
mv.Tm_Cve_Tipo_Movimiento, tm.Tm_Descripcion,
mv.Mv_Cantidad_1, mv.Mv_Cantidad_Control_1, mv.Mv_Unidad_Control_1,
mv.Mv_Tabla AS tipo_documento_origen,
mv.Mv_Documento AS documento_origen,
mv.Mv_Documento_Id,
mv.Mv_Referencia,
mv.Sc_Cve_Sucursal, mv.Al_Cve_Almacen,
mv.Es_Cve_Estado
FROM Movimiento mv
JOIN Sucursal sc ON sc.Sc_Cve_Sucursal = mv.Sc_Cve_Sucursal
JOIN Tipo_Movimiento tm ON tm.Tm_Cve_Tipo_Movimiento = mv.Tm_Cve_Tipo_Movimiento
JOIN Producto pr ON pr.Pr_Cve_Producto = mv.Pr_Cve_Producto
WHERE mv.Mv_Fecha >= '{fecha_ini}' AND mv.Mv_Fecha <= '{fecha_fin}'
AND sc.Em_Cve_Empresa = '0001'
AND mv.Tm_Cve_Tipo_Movimiento NOT IN ({excluidos_sql})
{filtro_cancelado}
""")
df["Pr_Cve_Producto"] = df["Pr_Cve_Producto"].astype(str).str.strip().str.zfill(PR_CVE_LEN)
clasif = df["Tm_Cve_Tipo_Movimiento"].map(CLASIFICACION)
es_tupla = clasif.map(lambda t: isinstance(t, tuple))
df["entrada_salida"] = [
clasif.iloc[i][0] if es_tupla.iloc[i] else "SIN CLASIFICAR" for i in range(len(df))
]
df["categoria_movimiento"] = [
clasif.iloc[i][1] if es_tupla.iloc[i] else f"SIN CLASIFICAR ({df['Tm_Descripcion'].iloc[i]})"
for i in range(len(df))
]
no_clasificados = sorted(
df.loc[~es_tupla, "Tm_Cve_Tipo_Movimiento"].unique().tolist()
)
if no_clasificados:
import warnings
warnings.warn(
f"Tm_Cve_Tipo_Movimiento sin clasificar en CLASIFICACION: {no_clasificados} "
f"-- aparecen como 'SIN CLASIFICAR' en el reporte. Agregarlos a movimientos_lib.py."
)
return df
def reconciliar(fecha_ini: str, fecha_fin: str, fecha_inicial_existencia: str, movimientos: pd.DataFrame):
"""Compara existencia_inicial + neto(movimientos) vs existencia_final (Metodo C).
Regresa (resumen_por_producto, productos_sin_cuadrar). El neto usa
Mv_Cantidad_Control_1 (misma unidad que Metodo C) -- NO Mv_Cantidad_1, ver decision
doc: para un puñado de productos divergen bastante y rompen la reconciliacion."""
ini = existencia_a_fecha_metodo_c(fecha_inicial_existencia).rename(
columns={"Mv_Cantidad_Control_1": "existencia_inicial"}
)
fin = existencia_a_fecha_metodo_c(fecha_fin).rename(
columns={"Mv_Cantidad_Control_1": "existencia_final"}
)
neto = movimientos.groupby("Pr_Cve_Producto", as_index=False)["Mv_Cantidad_Control_1"].sum().rename(
columns={"Mv_Cantidad_Control_1": "neto_movimientos"}
)
resumen = ini.merge(fin, on="Pr_Cve_Producto", how="outer").merge(
neto, on="Pr_Cve_Producto", how="outer"
)
for col in ("existencia_inicial", "existencia_final", "neto_movimientos"):
resumen[col] = resumen[col].fillna(0)
resumen["existencia_calculada"] = resumen["existencia_inicial"] + resumen["neto_movimientos"]
resumen["diferencia"] = resumen["existencia_calculada"] - resumen["existencia_final"]
resumen["cuadra"] = resumen["diferencia"].abs() <= TOLERANCIA
sin_cuadrar = resumen[~resumen["cuadra"]].copy()
return resumen, sin_cuadrar
def reconciliar_y_ajustar(fecha_ini: str, fecha_fin: str, movimientos: pd.DataFrame):
"""reconciliar() + reintento automatico reactivando cancelaciones/anulaciones SOLO
para los productos que no cuadraron -- ver decision doc. Regresa
(movimientos_final, resumen_final, sin_cuadrar_final)."""
fecha_inicial_existencia = (pd.Timestamp(fecha_ini) - pd.Timedelta(days=1)).strftime("%Y-%m-%d")
resumen, sin_cuadrar = reconciliar(fecha_ini, fecha_fin, fecha_inicial_existencia, movimientos)
if not len(sin_cuadrar):
return movimientos, resumen, sin_cuadrar
mov_con_anulaciones = extraer_movimientos(fecha_ini, fecha_fin, incluir_anulaciones=True)
productos_afectados = set(sin_cuadrar["Pr_Cve_Producto"])
mov_con_anulaciones_afectados = mov_con_anulaciones[
mov_con_anulaciones["Pr_Cve_Producto"].isin(productos_afectados)
]
mov_otros = movimientos[~movimientos["Pr_Cve_Producto"].isin(productos_afectados)]
mov_final = pd.concat([mov_otros, mov_con_anulaciones_afectados], ignore_index=True)
resumen2, sin_cuadrar2 = reconciliar(fecha_ini, fecha_fin, fecha_inicial_existencia, mov_final)
if len(sin_cuadrar2) < len(sin_cuadrar):
return mov_final, resumen2, sin_cuadrar2
return movimientos, resumen, sin_cuadrar
def reporte_completo(fecha_ini: str, fecha_fin: str):
"""Todo en uno: extrae movimientos, reconcilia, ajusta si hace falta. Regresa
(movimientos, resumen, sin_cuadrar)."""
mov = extraer_movimientos(fecha_ini, fecha_fin, incluir_anulaciones=False)
return reconciliar_y_ajustar(fecha_ini, fecha_fin, mov)