streamlit_app_movimientos.py
Ruta original: docs/proyectos/consumo-interno-fifo/scripts/streamlit_app_movimientos.py
"""
Movimientos en un periodo - reporte Streamlit (v1)
Movimientos efectivos de inventario (sin traspasos/transferencias internas, sin
cancelaciones/anulaciones salvo que hagan falta para cuadrar), reconciliados contra
existencia a una fecha (Metodo C, validado 100% contra MPRO). Ver
docs/schema/inventario.md (raiz de este repo) y el index.md de este proyecto.
Corre standalone con:
streamlit run streamlit_app_movimientos.py --server.port 8504
"""
import io
import os
import sys
from datetime import date
import pandas as pd
import streamlit as st
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, SCRIPT_DIR)
from movimientos_lib import reporte_completo # noqa: E402
VERSION = "v1"
def _is_standalone():
return not hasattr(st, "_layout_gastos_embedded")
def _mes_anterior_completo():
hoy = date.today()
primero_mes_actual = hoy.replace(day=1)
fin = primero_mes_actual - pd.Timedelta(days=1)
ini = fin.replace(day=1)
return ini, fin.date() if hasattr(fin, "date") else fin
@st.cache_data(ttl=1800, show_spinner="Consultando TRIVASADB3 (movimientos + existencia inicial/final)...")
def cargar_reporte(fecha_ini: str, fecha_fin: str):
mov, resumen, sin_cuadrar = reporte_completo(fecha_ini, fecha_fin)
return mov, resumen, sin_cuadrar
def fmt_num(x):
if x is None or pd.isna(x):
return "-"
return f"{x:,.2f}"
def render():
st.title("Movimientos en un periodo")
st.caption(f"Version {VERSION} · Inventario · reconciliado contra existencia a una fecha")
st.markdown(
"Movimientos efectivos de inventario en el periodo (compras, ventas, consumo interno, "
"producción, ajustes...) -- **sin** traspasos entre almacenes, transferencias entre "
"sucursales/en tránsito (netean a $0$, no son negocio), y sin cancelaciones/anulaciones "
"salvo que hagan falta para que el producto cuadre contra existencia inicial y final."
)
with st.sidebar:
st.subheader("Periodo")
default_ini, default_fin = _mes_anterior_completo()
fecha_ini = st.date_input("Desde", value=default_ini, key="mp_ini")
fecha_fin = st.date_input("Hasta", value=default_fin, key="mp_fin")
if fecha_ini > fecha_fin:
st.error("La fecha 'Desde' debe ser anterior o igual a 'Hasta'.")
return
mov, resumen, sin_cuadrar = cargar_reporte(str(fecha_ini), str(fecha_fin))
if mov.empty:
st.info("No hay movimientos para este periodo.")
return
# --- filtros (con los datos ya cargados, para que cambiar un filtro no dispare query) ---
with st.sidebar:
st.subheader("Filtros")
entrada_salida_sel = st.multiselect(
"Entrada / Salida", ["Entrada", "Salida"], default=["Entrada", "Salida"],
)
categorias = sorted(mov["categoria_movimiento"].dropna().unique().tolist())
categoria_sel = st.multiselect("Categoría de movimiento", categorias, default=[])
tipos_doc = sorted(mov["tipo_documento_origen"].dropna().unique().tolist())
tipo_doc_sel = st.multiselect("Tipo de documento origen", tipos_doc, default=[])
sucursales = sorted(mov["Sc_Cve_Sucursal"].dropna().unique().tolist())
sucursal_sel = st.multiselect("Sucursal", sucursales, default=[])
producto_buscar = st.text_input("Buscar producto (clave o descripción contiene)")
solo_no_cuadran = st.checkbox(
"Solo productos que NO cuadran contra existencia",
value=False,
help="Filtra la tabla de movimientos a solo los productos listados en la "
"pestaña Reconciliación como no cuadrados.",
)
df = mov.copy()
if entrada_salida_sel:
df = df[df["entrada_salida"].isin(entrada_salida_sel)]
if categoria_sel:
df = df[df["categoria_movimiento"].isin(categoria_sel)]
if tipo_doc_sel:
df = df[df["tipo_documento_origen"].isin(tipo_doc_sel)]
if sucursal_sel:
df = df[df["Sc_Cve_Sucursal"].isin(sucursal_sel)]
if producto_buscar:
needle = producto_buscar.strip().upper()
df = df[
df["Pr_Cve_Producto"].str.upper().str.contains(needle, na=False)
| df["Pr_Descripcion"].str.upper().str.contains(needle, na=False)
]
if solo_no_cuadran:
claves_no_cuadran = set(sin_cuadrar["Pr_Cve_Producto"])
df = df[df["Pr_Cve_Producto"].isin(claves_no_cuadran)]
entradas = df.loc[df["entrada_salida"] == "Entrada", "Mv_Cantidad_Control_1"].sum()
salidas = df.loc[df["entrada_salida"] == "Salida", "Mv_Cantidad_Control_1"].sum()
k1, k2, k3, k4, k5 = st.columns(5)
k1.metric("Filas (filtrado)", f"{len(df):,}")
k2.metric("Entradas (suma)", fmt_num(entradas))
k3.metric("Salidas (suma)", fmt_num(salidas))
k4.metric("Neto", fmt_num(entradas + salidas))
k5.metric(
"Productos que cuadran",
f"{len(resumen) - len(sin_cuadrar)} / {len(resumen)}",
help="Contra existencia a una fecha (Método C), sobre el universo completo del "
"periodo, no sobre lo filtrado en pantalla.",
)
tab_mov, tab_recon, tab_docs = st.tabs(["Movimientos", "Reconciliación", "Documentación"])
columnas_tabla = [
"Mv_Fecha", "Pr_Cve_Producto", "Pr_Descripcion", "entrada_salida",
"categoria_movimiento", "Mv_Cantidad_Control_1", "tipo_documento_origen",
"documento_origen", "Mv_Folio", "Sc_Cve_Sucursal", "Al_Cve_Almacen",
]
with tab_mov:
st.dataframe(
df[columnas_tabla].sort_values("Mv_Fecha", ascending=False),
width="stretch", height=520, hide_index=True,
)
st.caption(f"{len(df):,} filas de {len(mov):,} totales en el periodo")
buf = io.BytesIO()
with pd.ExcelWriter(buf, engine="openpyxl") as writer:
df[columnas_tabla].to_excel(writer, index=False, sheet_name="Movimientos")
st.download_button(
"Descargar Excel (filtrado)",
data=buf.getvalue(),
file_name=f"movimientos_{fecha_ini}_a_{fecha_fin}.xlsx",
mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
)
with tab_recon:
st.markdown(
"Por producto: `existencia_inicial` (Método C, día previo al periodo) + "
"`neto_movimientos` (de este listado) debe igualar `existencia_final` "
"(Método C, último día del periodo)."
)
st.dataframe(
resumen.sort_values("diferencia", key=abs, ascending=False),
width="stretch", height=420, hide_index=True,
)
if len(sin_cuadrar):
st.warning(f"{len(sin_cuadrar)} de {len(resumen)} productos no cuadran (tolerancia 0.5).")
else:
st.success(f"Los {len(resumen)} productos cuadran exactamente contra existencia inicial/final.")
with tab_docs:
st.markdown(
"Metodologia completa (clasificacion de tipos de movimiento, Metodo C de "
"existencia a una fecha, resultados de reconciliacion) en "
"`docs/schema/inventario.md` y `docs/proyectos/consumo-interno-fifo/index.md` "
"de este repo (trivasa-context)."
)
if _is_standalone():
st.set_page_config(page_title="Movimientos en un periodo", layout="wide")
render()