"""
Forecast Updater Service — auto-populate forecast actuals from connected data.
Mirrors the leak detection pattern: one function per forecast type, an
orchestrator that iterates companies, and a scheduler hook.
Sources:
Revenue actuals → QuickBooks invoices (Paid) > transactions (Payment)
Pipeline forecast → CRM deals weighted by probability
Cost actuals → QuickBooks expenses
"""
from __future__ import annotations
import logging
from datetime import datetime, timezone, timedelta
from typing import Any, Dict, List, Optional
from ..models import (
db,
Forecast,
Company,
Connector,
QuickbooksInvoice,
QuickbooksTransaction,
QuickbooksExpense,
CrmDeal,
)
logger = logging.getLogger(__name__)
# How many months ahead to auto-create future forecast periods
FUTURE_PERIODS_MONTHS = 6
# Minimum data points to trust a rolling average projection
MIN_HISTORY_MONTHS = 1
def _month_start(dt: Optional[datetime]) -> Optional[datetime]:
"""Floor a datetime to the first moment of its month (UTC)."""
if dt is None:
return None
return dt.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
def _month_end(dt: Optional[datetime]) -> Optional[datetime]:
"""Ceiling a datetime to the last moment of its month (UTC)."""
if dt is None:
return None
if dt.month == 12:
return dt.replace(year=dt.year + 1, month=1, day=1, hour=0, minute=0, second=0, microsecond=0) - timedelta(microseconds=1)
return dt.replace(month=dt.month + 1, day=1, hour=0, minute=0, second=0, microsecond=0) - timedelta(microseconds=1)
def _month_range(start: datetime) -> tuple:
"""Return (month_start, month_end) for a given datetime."""
return _month_start(start), _month_end(start)
# ---------------------------------------------------------------------------
# Revenue actuals — QuickBooks
# ---------------------------------------------------------------------------
def update_revenue_actual(
company_id: str,
period_start: datetime,
period_end: datetime,
) -> Dict[str, Any]:
"""Aggregate QB invoices/transactions for a period and write to Forecast.
Priority:
1. QuickbooksInvoice where status='Paid' and tx_date in period
2. QuickbooksTransaction where tx_type='Payment' and tx_date in period
"""
result: Dict[str, Any] = {
"company_id": company_id,
"period_start": period_start.isoformat(),
"period_end": period_end.isoformat(),
"source": None,
"actual_value": None,
"confidence": None,
"methodology": None,
"record_count": 0,
"updated": False,
}
# --- Try invoices first (Paid status, or empty status which QB treats as active) ---
invoices = (
QuickbooksInvoice.query
.filter_by(company_id=company_id)
.filter(
(QuickbooksInvoice.status == "Paid") |
(QuickbooksInvoice.status == "")
)
.filter(QuickbooksInvoice.tx_date >= period_start)
.filter(QuickbooksInvoice.tx_date <= period_end)
.filter(QuickbooksInvoice.total_amount.isnot(None))
.filter(QuickbooksInvoice.total_amount > 0)
.all()
)
if invoices:
total = sum((inv.total_amount or 0) for inv in invoices)
result["actual_value"] = round(total, 2)
result["source"] = "quickbooks_invoices"
result["confidence"] = 0.95
result["methodology"] = "quickbooks_invoices_paid"
result["record_count"] = len(invoices)
else:
# --- Fallback to transactions ---
transactions = (
QuickbooksTransaction.query
.filter_by(company_id=company_id)
.filter(QuickbooksTransaction.tx_type == "Payment")
.filter(QuickbooksTransaction.tx_date >= period_start)
.filter(QuickbooksTransaction.tx_date <= period_end)
.all()
)
if transactions:
total = sum((txn.amount or 0) for txn in transactions)
result["actual_value"] = round(total, 2)
result["source"] = "quickbooks_transactions"
result["confidence"] = 0.8
result["methodology"] = "quickbooks_transactions_payment"
result["record_count"] = len(transactions)
# --- Write to Forecast if we found data ---
if result["actual_value"] is not None:
forecast = Forecast.query.filter_by(
company_id=company_id,
forecast_type="revenue",
period="monthly",
period_start=period_start,
period_end=period_end,
).first()
if forecast:
forecast.actual_value = result["actual_value"]
forecast.confidence = result["confidence"]
forecast.methodology = result["methodology"]
forecast.assumptions_json = {
"source": result["source"],
"record_count": result["record_count"],
"updated_by": "forecast_updater",
}
db.session.commit()
result["updated"] = True
logger.info(
"Revenue actual updated: company=%s period=%s actual=$%.2f (%s, %d records)",
company_id[:8], period_start.strftime("%Y-%m"),
result["actual_value"], result["source"], result["record_count"],
)
return result
# ---------------------------------------------------------------------------
# Pipeline forecast — CRM deals weighted by probability
# ---------------------------------------------------------------------------
def update_pipeline_forecast(
company_id: str,
period_start: datetime,
period_end: datetime,
) -> Dict[str, Any]:
"""Aggregate open CRM deals expected to close in the period.
Weighted sum: SUM(amount * probability), excluding closed/lost deals.
"""
result: Dict[str, Any] = {
"company_id": company_id,
"period_start": period_start.isoformat(),
"period_end": period_end.isoformat(),
"source": "crm_deals",
"projected_value": None,
"confidence": None,
"methodology": None,
"deal_count": 0,
"updated": False,
}
deals = (
CrmDeal.query
.filter_by(company_id=company_id)
.filter(CrmDeal.stage != "closedlost")
.filter(CrmDeal.expected_close_date >= period_start)
.filter(CrmDeal.expected_close_date <= period_end)
.all()
)
if not deals:
return result
weighted_sum = 0.0
deal_count = 0
for deal in deals:
if deal.amount and deal.probability is not None:
weighted_sum += deal.amount * deal.probability
deal_count += 1
result["projected_value"] = round(weighted_sum, 2)
result["deal_count"] = deal_count
result["confidence"] = min(0.95, 0.5 + 0.1 * deal_count) # More deals → more confidence
result["methodology"] = "weighted_pipeline"
# --- Upsert forecast record ---
forecast = Forecast.query.filter_by(
company_id=company_id,
forecast_type="pipeline",
period="monthly",
period_start=period_start,
period_end=period_end,
).first()
if forecast:
forecast.projected_value = result["projected_value"]
forecast.confidence = result["confidence"]
forecast.methodology = result["methodology"]
forecast.assumptions_json = {
"source": "crm_deals",
"deal_count": deal_count,
"total_deals_considered": len(deals),
"methodology": "weighted_by_probability",
"updated_by": "forecast_updater",
}
db.session.commit()
result["updated"] = True
else:
forecast = Forecast(
company_id=company_id,
forecast_type="pipeline",
period="monthly",
period_start=period_start,
period_end=period_end,
projected_value=result["projected_value"],
confidence=result["confidence"],
methodology=result["methodology"],
assumptions_json={
"source": "crm_deals",
"deal_count": deal_count,
"total_deals_considered": len(deals),
"methodology": "weighted_by_probability",
"auto_created": True,
"updated_by": "forecast_updater",
},
)
db.session.add(forecast)
db.session.commit()
result["updated"] = True
result["created"] = True
logger.info(
"Pipeline forecast updated: company=%s period=%s projected=$%.2f (%d deals)",
company_id[:8], period_start.strftime("%Y-%m"),
result["projected_value"], result["deal_count"],
)
return result
# ---------------------------------------------------------------------------
# Cost actuals — QuickBooks expenses
# ---------------------------------------------------------------------------
def update_cost_actual(
company_id: str,
period_start: datetime,
period_end: datetime,
) -> Dict[str, Any]:
"""Aggregate QB expenses for a period and write to Forecast."""
result: Dict[str, Any] = {
"company_id": company_id,
"period_start": period_start.isoformat(),
"period_end": period_end.isoformat(),
"source": None,
"actual_value": None,
"confidence": None,
"methodology": None,
"record_count": 0,
"updated": False,
}
expenses = (
QuickbooksExpense.query
.filter_by(company_id=company_id)
.filter(QuickbooksExpense.tx_date >= period_start)
.filter(QuickbooksExpense.tx_date <= period_end)
.filter(QuickbooksExpense.amount.isnot(None))
.filter(QuickbooksExpense.amount > 0)
.all()
)
if not expenses:
return result
total = sum((exp.amount or 0) for exp in expenses)
result["actual_value"] = round(total, 2)
result["source"] = "quickbooks_expenses"
result["confidence"] = 0.9
result["methodology"] = "quickbooks_expenses"
result["record_count"] = len(expenses)
forecast = Forecast.query.filter_by(
company_id=company_id,
forecast_type="cost",
period="monthly",
period_start=period_start,
period_end=period_end,
).first()
if forecast:
forecast.actual_value = result["actual_value"]
forecast.confidence = result["confidence"]
forecast.methodology = result["methodology"]
forecast.assumptions_json = {
"source": result["source"],
"record_count": result["record_count"],
"updated_by": "forecast_updater",
}
db.session.commit()
result["updated"] = True
else:
# Create cost forecast if it doesn't exist
forecast = Forecast(
company_id=company_id,
forecast_type="cost",
period="monthly",
period_start=period_start,
period_end=period_end,
projected_value=0.0,
actual_value=result["actual_value"],
confidence=result["confidence"],
methodology=result["methodology"],
assumptions_json={
"source": result["source"],
"record_count": result["record_count"],
"auto_created": True,
"updated_by": "forecast_updater",
},
)
db.session.add(forecast)
db.session.commit()
result["updated"] = True
result["created"] = True
logger.info(
"Cost actual updated: company=%s period=%s actual=$%.2f (%d expenses)",
company_id[:8], period_start.strftime("%Y-%m"),
result["actual_value"], result["record_count"],
)
return result
# ---------------------------------------------------------------------------
# Auto-create future forecast periods
# ---------------------------------------------------------------------------
def _calculate_projected_from_history(
company_id: str, forecast_type: str
) -> Optional[float]:
"""Calculate projected value from the rolling average of past actuals.
Uses the last 3 months of actual data as a baseline.
"""
now = datetime.now(timezone.utc)
three_months_ago = now - timedelta(days=90)
# Get recent actuals
recent_forecasts = (
Forecast.query
.filter_by(company_id=company_id, forecast_type=forecast_type, period="monthly")
.filter(Forecast.period_start >= three_months_ago)
.filter(Forecast.actual_value.isnot(None))
.order_by(Forecast.period_start)
.all()
)
if len(recent_forecasts) < MIN_HISTORY_MONTHS:
return None
total = sum(f.actual_value or 0 for f in recent_forecasts)
return round(total / len(recent_forecasts), 2)
def auto_create_future_forecasts(
company_id: str,
forecast_types: Optional[List[str]] = None,
) -> Dict[str, Any]:
"""Auto-create forecast periods for the next N months if they don't exist.
Uses rolling average of past actuals as the projected baseline.
"""
if forecast_types is None:
forecast_types = ["revenue"]
result: Dict[str, Any] = {
"company_id": company_id,
"created": [],
"skipped": [],
"total_created": 0,
}
now = datetime.now(timezone.utc)
current_month_start = _month_start(now)
for forecast_type in forecast_types:
# Calculate baseline from history
projected = _calculate_projected_from_history(company_id, forecast_type)
for offset in range(1, FUTURE_PERIODS_MONTHS + 1):
future_start = current_month_start + timedelta(days=31 * offset)
future_start = _month_start(future_start)
future_end = _month_end(future_start)
# Skip if already exists
existing = Forecast.query.filter_by(
company_id=company_id,
forecast_type=forecast_type,
period="monthly",
period_start=future_start,
period_end=future_end,
).first()
if existing:
result["skipped"].append({
"forecast_type": forecast_type,
"period_start": future_start.isoformat(),
"reason": "already_exists",
})
continue
# Determine projected value
projected_value = projected
confidence = 0.6 if projected is not None else 0.3
methodology = "rolling_3_month_average" if projected is not None else "default"
forecast = Forecast(
company_id=company_id,
forecast_type=forecast_type,
period="monthly",
period_start=future_start,
period_end=future_end,
projected_value=projected_value or 0.0,
confidence=confidence,
methodology=methodology,
assumptions_json={
"auto_created": True,
"baseline_source": methodology,
"history_months_used": 3 if projected else 0,
"updated_by": "forecast_updater",
},
)
db.session.add(forecast)
result["created"].append({
"forecast_type": forecast_type,
"period_start": future_start.strftime("%Y-%m"),
"projected_value": projected_value,
})
result["total_created"] += 1
db.session.commit()
logger.info(
"Auto-created %d future forecast(s) for company %s",
result["total_created"], company_id[:8],
)
return result
# ---------------------------------------------------------------------------
# Orchestrator — sync all forecasts for a company
# ---------------------------------------------------------------------------
def sync_all_forecasts(company_id: str) -> Dict[str, Any]:
"""Sync all forecast types for a company.
Updates actuals for past/current months, creates future periods,
and populates pipeline from CRM deals.
Returns a summary dict matching the leak detection pattern.
"""
result: Dict[str, Any] = {
"company_id": company_id,
"revenue_updated": 0,
"pipeline_updated": 0,
"cost_updated": 0,
"future_created": 0,
"errors": [],
"duration_seconds": 0,
}
import time
start_time = time.monotonic()
try:
# Check what connectors are active AND whether QB data exists directly
qb_connector = Connector.query.filter_by(
company_id=company_id, service="quickbooks", status="connected"
).first()
crm_connector = Connector.query.filter_by(
company_id=company_id, service="hubspot", status="connected"
).first()
has_qb = qb_connector is not None
has_crm = crm_connector is not None
# Fallback: if no QB connector but QB data exists, treat as connected
if not has_qb:
qb_inv_count = QuickbooksInvoice.query.filter_by(company_id=company_id).count()
qb_txn_count = QuickbooksTransaction.query.filter_by(company_id=company_id).count()
qb_exp_count = QuickbooksExpense.query.filter_by(company_id=company_id).count()
if qb_inv_count > 0 or qb_txn_count > 0 or qb_exp_count > 0:
has_qb = True
logger.info(
"No QB connector for company %s, but found data (%d inv, %d txn, %d exp) — syncing",
company_id[:8], qb_inv_count, qb_txn_count, qb_exp_count,
)
# Fallback: if no CRM connector but deals exist, treat as connected
if not has_crm:
crm_count = CrmDeal.query.filter_by(company_id=company_id).count()
if crm_count > 0:
has_crm = True
logger.info(
"No CRM connector for company %s, but found %d deals — syncing",
company_id[:8], crm_count,
)
if not has_qb and not has_crm:
logger.info(
"No QB or CRM connectors for company %s — skipping forecast sync",
company_id[:8],
)
return result
now = datetime.now(timezone.utc)
current_month_start, current_month_end = _month_range(now)
# Existing forecast periods are naive datetimes — make ours naive too for comparison
naive_current_start = current_month_start.replace(tzinfo=None)
# Get existing revenue forecasts to determine which periods to update
revenue_forecasts = Forecast.query.filter_by(
company_id=company_id,
forecast_type="revenue",
period="monthly",
).order_by(Forecast.period_start).all()
# Update actuals for past + current months
if has_qb:
for forecast in revenue_forecasts:
if forecast.period_start <= naive_current_start:
r = update_revenue_actual(
company_id, forecast.period_start, forecast.period_end
)
if r["updated"]:
result["revenue_updated"] += 1
# Update cost actuals
cost_forecasts = Forecast.query.filter_by(
company_id=company_id,
forecast_type="cost",
period="monthly",
).order_by(Forecast.period_start).all()
for forecast in cost_forecasts:
if forecast.period_start <= naive_current_start:
c = update_cost_actual(
company_id, forecast.period_start, forecast.period_end
)
if c["updated"]:
result["cost_updated"] += 1
# Update pipeline forecasts
if has_crm:
for offset in range(-3, FUTURE_PERIODS_MONTHS + 1):
period_start = current_month_start + timedelta(days=31 * offset)
period_start = _month_start(period_start)
period_end = _month_end(period_start)
p = update_pipeline_forecast(company_id, period_start, period_end)
if p["updated"]:
result["pipeline_updated"] += 1
# Auto-create future periods
forecast_types_to_create = ["revenue"]
if has_qb:
forecast_types_to_create.append("cost")
future_result = auto_create_future_forecasts(company_id, forecast_types_to_create)
result["future_created"] = future_result["total_created"]
except Exception:
logger.exception("Forecast sync failed for company %s", company_id[:8])
result["errors"].append(str(db.session.rollback()))
result["duration_seconds"] = round(time.monotonic() - start_time, 3)
return result
def run_all_forecasts() -> Dict[str, Any]:
"""Run forecast sync for all active companies (scheduler entry point)."""
logger.info("Running scheduled forecast update for all companies...")
result: Dict[str, Any] = {
"companies_processed": 0,
"total_revenue_updated": 0,
"total_pipeline_updated": 0,
"total_cost_updated": 0,
"total_future_created": 0,
"errors": [],
}
companies = Company.query.filter_by(is_deleted=False).all()
for company in companies:
try:
r = sync_all_forecasts(company.id)
result["companies_processed"] += 1
result["total_revenue_updated"] += r["revenue_updated"]
result["total_pipeline_updated"] += r["pipeline_updated"]
result["total_cost_updated"] += r["cost_updated"]
result["total_future_created"] += r["future_created"]
if r["errors"]:
result["errors"].append({
"company_id": company.id[:8],
"errors": r["errors"],
})
except Exception:
logger.exception("Forecast sync failed for company %s", company.id[:8])
result["errors"].append({
"company_id": company.id[:8],
"error": "unexpected_error",
})
logger.info(
"Forecast scan complete: %d companies, %d revenue, %d pipeline, "
"%d cost, %d future created",
result["companies_processed"],
result["total_revenue_updated"],
result["total_pipeline_updated"],
result["total_cost_updated"],
result["total_future_created"],
)
return result