La valoración de posiciones es el motor analítico que determina el valor económico futuro de cada ubicación geográfica, permitiendo a SIGA tomar decisiones óptimas sobre dónde posicionar los activos de la flota.
"Una posición no vale por sí misma, sino por las oportunidades futuras que genera"
class PositionValue:
"""Modelo completo de valoración de una posición geográfica"""
def __init__(self, zone_id: str, timestamp: datetime):
self.zone_id = zone_id
self.timestamp = timestamp
# Componentes del valor
self.immediate_demand = 0.0 # Demanda en las próximas 4h
self.short_term_demand = 0.0 # Demanda 4-24h
self.medium_term_demand = 0.0 # Demanda 24-72h
self.competition_factor = 1.0 # 0-1, menos es mejor
self.accessibility_score = 0.0 # Facilidad de acceso a otras zonas
self.operational_cost = 0.0 # Costo de operar en la zona
self.seasonality_factor = 1.0 # Ajuste estacional
self.special_events_factor = 1.0 # Eventos especiales
def calculate_total_value(self) -> float:
"""Calcula el valor total ponderado de la posición"""
# Pesos por horizonte temporal
weights = {
'immediate': 0.4,
'short_term': 0.35,
'medium_term': 0.25
}
# Valor base de demanda
demand_value = (
self.immediate_demand * weights['immediate'] +
self.short_term_demand * weights['short_term'] +
self.medium_term_demand * weights['medium_term']
)
# Ajustes
adjusted_value = demand_value * (
(2 - self.competition_factor) * # Bonus por poca competencia
self.seasonality_factor *
self.special_events_factor
)
# Valor neto considerando costos
net_value = adjusted_value - self.operational_cost
# Bonus por accesibilidad (posición estratégica)
strategic_value = net_value + (self.accessibility_score * 0.1)
return max(0, strategic_value)
def calculate_zone_demand(zone_id: str, time_window: TimeWindow) -> DemandProfile:
"""Calcula el perfil de demanda para una zona específica"""
demand = DemandProfile()
# Análisis histórico
historical_loads = get_historical_loads(
zone_id=zone_id,
days_back=90,
time_of_day=time_window.start.hour,
day_of_week=time_window.start.weekday()
)
# Estadísticas base
demand.average_loads_per_day = historical_loads.mean()
demand.load_variance = historical_loads.std()
demand.peak_hours = identify_peak_hours(historical_loads)
# Tendencias
demand.growth_trend = calculate_trend(historical_loads)
demand.weekly_pattern = extract_weekly_pattern(historical_loads)
# Predicción ML
features = prepare_features(zone_id, time_window)
demand.ml_prediction = ml_model.predict(features)
# Confianza en la predicción
demand.confidence = calculate_prediction_confidence(
historical_variance=demand.load_variance,
data_points=len(historical_loads),
model_accuracy=ml_model.accuracy
)
return demand
def calculate_competition_factor(zone_id: str, time_window: TimeWindow) -> float:
"""Evalúa el nivel de competencia en una zona"""
# Vehículos actuales en la zona
current_vehicles = count_vehicles_in_zone(zone_id)
# Vehículos en ruta hacia la zona
incoming_vehicles = count_vehicles_heading_to_zone(zone_id, time_window)
# Capacidad total disponible
total_capacity = calculate_total_capacity(
current_vehicles + incoming_vehicles
)
# Demanda esperada
expected_demand = predict_zone_demand(zone_id, time_window)
# Ratio oferta/demanda
supply_demand_ratio = total_capacity / max(expected_demand, 1)
# Factor de competencia (0 = sin competencia, 1 = saturado)
if supply_demand_ratio < 0.5:
return 0.0 # Demanda muy superior a oferta
elif supply_demand_ratio < 1.0:
return 0.3 # Balance favorable
elif supply_demand_ratio < 1.5:
return 0.6 # Competencia moderada
elif supply_demand_ratio < 2.0:
return 0.8 # Alta competencia
else:
return 1.0 # Saturación
def calculate_accessibility_score(zone_id: str) -> float:
"""Evalúa qué tan bien conectada está una zona"""
score_components = []
# Conexiones directas a zonas de alta demanda
high_demand_zones = get_high_demand_zones()
for hdz in high_demand_zones:
distance = calculate_distance(zone_id, hdz.id)
if distance < 150: # Radio relevante
connection_value = hdz.average_demand / (distance + 1)
score_components.append(connection_value)
# Proximidad a corredores principales
main_corridors = get_main_transport_corridors()
corridor_distance = min_distance_to_corridors(zone_id, main_corridors)
corridor_score = 100 / (corridor_distance + 10)
score_components.append(corridor_score)
# Facilidades logísticas
facilities = get_logistics_facilities(zone_id)
facility_score = len(facilities) * 20 # Bonus por cada facilidad
score_components.append(facility_score)
# Score total normalizado
return sum(score_components) / len(score_components)
┌─────────────────────────────────────────────────────────────┐
│ MATRIZ DE VALOR - HORA 14:00 │
├─────────────┬───────────┬──────────┬───────────┬───────────┤
│ ZONA │ DEMANDA │ COMPET. │ ACCESO │ VALOR │
│ │ (€/24h) │ FACTOR │ SCORE │ TOTAL │
├─────────────┼───────────┼──────────┼───────────┼───────────┤
│ Barcelona │ 4,500 │ 0.7 │ 95 │ 3,285 │
│ Valencia │ 3,200 │ 0.8 │ 88 │ 2,160 │
│ Zaragoza │ 2,800 │ 0.4 │ 92 │ 2,436 │
│ Madrid │ 5,100 │ 0.9 │ 98 │ 3,069 │
│ Alicante │ 1,900 │ 0.3 │ 72 │ 1,729 │
│ Murcia │ 2,100 │ 0.5 │ 78 │ 1,638 │
└─────────────┴───────────┴──────────┴───────────┴───────────┘
🔥 Zonas Calientes (Alta Valor): Barcelona, Madrid
⚠️ Zonas Equilibradas: Zaragoza, Valencia
❄️ Zonas Frías: Alicante, Murcia
class DynamicPositionValuation:
"""Sistema de valoración que se actualiza en tiempo real"""
def __init__(self):
self.value_cache = {}
self.last_update = {}
self.update_threshold = timedelta(minutes=15)
def get_position_value(self, zone_id: str,
timestamp: datetime) -> PositionValue:
"""Obtiene el valor actualizado de una posición"""
# Check cache
if self._is_cache_valid(zone_id, timestamp):
return self.value_cache[zone_id]
# Recalcular
value = self._calculate_fresh_value(zone_id, timestamp)
# Actualizar cache
self.value_cache[zone_id] = value
self.last_update[zone_id] = timestamp
return value
def _calculate_fresh_value(self, zone_id: str,
timestamp: datetime) -> PositionValue:
"""Calcula el valor fresco de una posición"""
value = PositionValue(zone_id, timestamp)
# Componentes en paralelo para performance
with ThreadPoolExecutor(max_workers=5) as executor:
futures = {
'demand': executor.submit(
self._calculate_demand_component, zone_id, timestamp
),
'competition': executor.submit(
self._calculate_competition_component, zone_id, timestamp
),
'accessibility': executor.submit(
self._calculate_accessibility_component, zone_id
),
'costs': executor.submit(
self._calculate_cost_component, zone_id
),
'events': executor.submit(
self._check_special_events, zone_id, timestamp
)
}
# Recopilar resultados
results = {k: future.result() for k, future in futures.items()}
# Aplicar al modelo
value.immediate_demand = results['demand']['immediate']
value.short_term_demand = results['demand']['short_term']
value.medium_term_demand = results['demand']['medium_term']
value.competition_factor = results['competition']
value.accessibility_score = results['accessibility']
value.operational_cost = results['costs']
value.special_events_factor = results['events']
return value
def generate_value_heatmap(timestamp: datetime) -> HeatmapData:
"""Genera datos para visualización de mapa de calor"""
heatmap = HeatmapData()
# Obtener todas las zonas
zones = get_all_zones()
for zone in zones:
# Calcular valor actual
value = valuation_engine.get_position_value(zone.id, timestamp)
# Datos para visualización
heatmap.add_point(
lat=zone.center_lat,
lng=zone.center_lng,
value=value.calculate_total_value(),
radius=zone.radius_km,
metadata={
'zone_name': zone.name,
'immediate_demand': value.immediate_demand,
'competition': value.competition_factor,
'trend': calculate_value_trend(zone.id, hours=24)
}
)
# Gradiente de colores
heatmap.set_gradient({
0.0: '#0000FF', # Azul - Valor bajo
0.25: '#00FFFF', # Cyan
0.5: '#00FF00', # Verde
0.75: '#FFFF00', # Amarillo
1.0: '#FF0000' # Rojo - Valor alto
})
return heatmap
def adjust_post_holiday_values(base_value: PositionValue,
holiday_info: HolidayInfo) -> PositionValue:
"""Ajusta valores después de días festivos"""
adjusted = copy.deepcopy(base_value)
if holiday_info.is_national_holiday:
# Día después de festivo nacional
if holiday_info.days_since_holiday == 1:
adjusted.immediate_demand *= 1.4 # Pico de demanda
adjusted.competition_factor *= 0.7 # Menos competencia
elif holiday_info.days_since_holiday == 2:
adjusted.immediate_demand *= 1.2
adjusted.short_term_demand *= 1.1
elif holiday_info.is_regional_holiday:
# Ajustes regionales
if base_value.zone_id in holiday_info.affected_zones:
adjusted.immediate_demand *= 1.25
return adjusted
def detect_special_events(zone_id: str, date_range: DateRange) -> List[Event]:
"""Detecta eventos que afectan el valor de una zona"""
events = []
# Eventos planificados
scheduled_events = query_events_database(zone_id, date_range)
events.extend(scheduled_events)
# Patrones históricos (ferias, congresos recurrentes)
historical_patterns = detect_recurring_events(zone_id, date_range)
events.extend(historical_patterns)
# Eventos detectados por anomalías
anomaly_events = detect_demand_anomalies(zone_id, date_range)
events.extend(anomaly_events)
return events
| Métrica | Target | Actual |
|---|---|---|
| Precisión valoración | >85% | 87.3% |
| Tiempo cálculo | <500ms | 420ms |
| Cache hit rate | >90% | 92.1% |
| Cobertura zonas | 100% | 100% |
class MLValuePredictor:
"""Predictor ML para valores de posición"""
def __init__(self):
self.model = self._load_trained_model()
self.feature_encoder = FeatureEncoder()
def predict_value_evolution(self, zone_id: str,
horizon_hours: int) -> ValuePrediction:
"""Predice la evolución del valor en las próximas horas"""
# Preparar features
features = self.feature_encoder.encode({
'zone_id': zone_id,
'hour_of_day': datetime.now().hour,
'day_of_week': datetime.now().weekday(),
'historical_demand': get_historical_demand_stats(zone_id),
'weather_forecast': get_weather_forecast(zone_id),
'economic_indicators': get_economic_indicators(),
'competitor_positions': get_competitor_analysis(zone_id)
})
# Predicción
predictions = []
for hour in range(horizon_hours):
pred = self.model.predict(features, hour_offset=hour)
predictions.append({
'hour': hour,
'value': pred.value,
'confidence': pred.confidence,
'factors': pred.contributing_factors
})
return ValuePrediction(predictions)
⬅️ Volver a Procesos Core | ➡️ Siguiente: Predicción de Demanda