El sistema SIGA est谩 construido con una arquitectura de microservicios donde cada componente tiene una responsabilidad espec铆fica y bien definida. Esta modularidad permite escalabilidad independiente y mantenimiento eficiente.
Coraz贸n del sistema que eval煤a opciones y genera recomendaciones inteligentes para cada situaci贸n operativa.
class DecisionEngine:
"""Motor principal de toma de decisiones"""
def evaluate_options(self, context: DecisionContext) -> List[Option]:
"""Eval煤a todas las opciones disponibles"""
# 1. Obtener opciones posibles
# 2. Calcular valor de cada opci贸n
# 3. Considerar restricciones
# 4. Rankear por valor esperado
def calculate_opportunity_cost(self, decision: Decision) -> float:
"""Calcula el costo de oportunidad de cada decisi贸n"""
def generate_recommendation(self, options: List[Option]) -> Recommendation:
"""Genera recomendaci贸n con justificaci贸n"""
/api/v1/decisionsPredice demanda futura, tiempos de espera y probabilidades de carga usando Machine Learning.
frameworks:
- scikit-learn: Modelos base
- XGBoost: Modelos principales
- TensorFlow: Experimentaci贸n
- Prophet: Series temporales
infrastructure:
- MLflow: Gesti贸n de modelos
- Airflow: Pipelines de entrenamiento
- S3: Almacenamiento de modelos
class ZoneDemandPredictor:
"""Predice demanda por zona geogr谩fica"""
features = [
'zone_id', 'day_of_week', 'hour_of_day',
'historical_demand_7d', 'historical_demand_30d',
'weather_forecast', 'local_events',
'competitor_activity', 'seasonality'
]
def predict(self, zone: str, horizon: int) -> DemandForecast:
# Returns probability distribution of demand
pass
class WaitTimeEstimator:
"""Estima tiempo hasta pr贸xima carga"""
def estimate(self, location: Location, vehicle_type: str) -> WaitTimePrediction:
# Returns expected wait time with confidence intervals
pass
class PositionValuePredictor:
"""Calcula el valor futuro de estar en una posici贸n"""
def predict_value(self, position: Location, time: datetime) -> float:
# Returns expected monetary value
pass
Resuelve problemas complejos de optimizaci贸n multi-objetivo considerando toda la flota.
def solve_vrp_time_windows(
vehicles: List[Vehicle],
orders: List[Order],
time_windows: List[TimeWindow]
) -> Solution:
"""Resuelve Vehicle Routing Problem con restricciones temporales"""
# Create routing model
manager = pywrapcp.RoutingIndexManager(...)
routing = pywrapcp.RoutingModel(manager)
# Add constraints
# - Capacity
# - Time windows
# - Driver hours
# - Vehicle compatibility
# Solve
solution = routing.SolveWithParameters(search_parameters)
return parse_solution(solution)
def optimize_multi_day_planning(
fleet: Fleet,
demand_forecast: DemandForecast,
horizon_days: int = 3
) -> MultiDayPlan:
"""Optimiza posicionamiento considerando demanda futura"""
# Objective: Maximize total value over horizon
# Constraints: Operational feasibility
# Output: Positioning strategy
Mantiene el estado en tiempo real de toda la flota y genera eventos de posici贸n.
components:
- GPS Gateway: Recibe datos de dispositivos
- Stream Processor: Procesa eventos en tiempo real
- State Store: Mantiene estado actual
- Event Publisher: Distribuye actualizaciones
interface VehiclePosition {
vehicleId: string;
timestamp: Date;
location: {
lat: number;
lng: number;
};
speed: number;
heading: number;
status: VehicleStatus;
driver: DriverInfo;
cargo: CargoInfo;
}
Centraliza todas las integraciones con sistemas externos manteniendo adaptadores espec铆ficos.
class RtracktorAdapter:
"""Integraci贸n con sistema Rtracktor"""
async def get_available_loads(self) -> List[Load]:
"""Obtiene cargas disponibles"""
async def update_delivery_status(self, delivery_id: str, status: str):
"""Actualiza estado de entrega"""
async def get_vehicle_info(self, vehicle_id: str) -> Vehicle:
"""Obtiene informaci贸n del veh铆culo"""
class MapsService:
"""Integraci贸n con servicios de mapas"""
providers = {
'routing': 'GraphHopper',
'geocoding': 'Nominatim',
'traffic': 'TomTom'
}
async def calculate_route(self, origin: Location, destination: Location) -> Route:
"""Calcula ruta 贸ptima considerando tr谩fico"""
class WeatherService:
"""Integraci贸n con servicios meteorol贸gicos"""
async def get_forecast(self, location: Location, hours: int) -> WeatherForecast:
"""Obtiene predicci贸n meteorol贸gica"""
Gestiona todas las comunicaciones y alertas del sistema hacia usuarios.
class NotificationTypes(Enum):
# Operativas
NEW_LOAD_AVAILABLE = "new_load"
DECISION_REQUIRED = "decision_required"
ROUTE_OPTIMIZED = "route_optimized"
# Alertas
VEHICLE_DELAYED = "vehicle_delayed"
DEMAND_SPIKE = "demand_spike"
FLEET_IMBALANCE = "fleet_imbalance"
# Sistema
PREDICTION_READY = "prediction_ready"
MAINTENANCE_DUE = "maintenance_due"
Procesa datos hist贸ricos para generar insights y alimentar dashboards.
data_pipeline:
- Apache Spark: Procesamiento batch
- Kafka: Streaming analytics
- Elasticsearch: B煤squeda y agregaci贸n
- Grafana: Visualizaci贸n
analytics_jobs:
- daily_performance_metrics
- weekly_pattern_analysis
- monthly_optimization_report
- realtime_kpi_calculation
-- Ejemplo: C谩lculo de eficiencia de flota
WITH fleet_metrics AS (
SELECT
date_trunc('day', timestamp) as day,
vehicle_id,
SUM(CASE WHEN cargo_weight > 0 THEN distance_km ELSE 0 END) as loaded_km,
SUM(CASE WHEN cargo_weight = 0 THEN distance_km ELSE 0 END) as empty_km,
COUNT(DISTINCT order_id) as deliveries_completed
FROM vehicle_movements
GROUP BY 1, 2
)
SELECT
day,
AVG(loaded_km / NULLIF(loaded_km + empty_km, 0)) as utilization_rate,
SUM(loaded_km + empty_km) as total_km,
AVG(deliveries_completed) as avg_deliveries_per_vehicle
FROM fleet_metrics
GROUP BY day;
Punto de entrada 煤nico para todas las peticiones externas, gestiona autenticaci贸n, rate limiting y routing.
routes:
- path: /api/v1/decisions/*
service: decision-engine
rate_limit: 100/min
auth: jwt
- path: /api/v1/predictions/*
service: prediction-service
rate_limit: 50/min
auth: jwt
- path: /api/v1/fleet/*
service: fleet-tracking
rate_limit: 1000/min
auth: jwt
- path: /ws/real-time
service: websocket-server
protocol: websocket
auth: jwt
security:
- jwt_validation
- api_key_backup
- ip_whitelist
- cors_policy
health_endpoints = {
'decision-engine': '/health',
'prediction-service': '/health',
'optimization-service': '/health',
'fleet-tracking': '/health',
'integration-hub': '/health',
'notification-service': '/health',
'analytics-engine': '/health',
'api-gateway': '/health'
}
| Componente | M茅trica Clave | Target | Alerta |
|---|---|---|---|
| Decision Engine | Latencia p99 | <500ms | >1s |
| Prediction Service | Accuracy | >85% | <80% |
| Optimization Service | Tiempo soluci贸n | <3s | >5s |
| Fleet Tracking | Lag de datos | <5s | >30s |
| Integration Hub | Disponibilidad | 99.9% | <99% |
| Notification Service | Delivery rate | >98% | <95% |
| Analytics Engine | Job completion | 100% | Failed |
| API Gateway | Error rate | <0.1% | >1% |