Time series forecasting and anomaly detection

The following steps are involved in forecasting/anomaly detection:

This model allows you to forecast a future value based on previous ones with a random forest. Consider a time series consisting of N data points, represented as (t_i, v_i) for i = 1, 2, …, N, where t_i denotes the timestamp and v_i is the corresponding observed value.

Forecasting: The objective is to predict future values of the time series at upcoming timestamps t_N+1, t_N+2, and so on. These predictions are based on patterns learned from the existing data.

Anomaly Detection: When the actual value at a future timestamp (e.g., v_N+1) becomes available, it can be compared with the predicted value. If the difference between the actual and predicted values exceeds a defined threshold, the data point is flagged as an anomaly.

Seasonality Decomposition: If the time series exhibits recurring seasonal patterns (e.g., temperature rises at noon each day), we can decompose the time series to isolate the seasonal component. The deseasonalized data can then be used as input to forecasting models such as Random Forest.

Viewing and interpreting anomaly detection results

After running your real-time forecasting and anomaly detection pipeline, you can visualize the results using a time series chart. This visualization helps you compare actual values to predicted values and highlights anomalies based on deviation thresholds.

The chart typically includes:

• Actual values: the raw observed data points over time.

• Predicted values: values forecasted by the Foreseer AI agent using historical trends and time series models such as statistical or machine learning algorithms.

• Anomalies: points where the actual value significantly diverges from the prediction, exceeding a statistical error threshold.

The following example visualization illustrates how anomalies (red markers) are detected where the actual time series deviates beyond the expected prediction range:

This graphical output helps validate the accuracy of the model and supports use cases such as system monitoring, fraud detection, or infrastructure anomaly tracking.

      SELECT forecast(<Double field name>, <DateTime field name>, <int>, [ <JsonNode> ])
      FROM <stream name>
    

      SELECT forecast(V, TS, 60) FROM S;
      SELECT forecast(V, null, 1) FROM S;
    

      ti+1 = ti + 60 seconds
    

Stream S → Jumping Window W → Forecast CQ → Output S

      SELECT forecast(opNum, opTime, 60, makeJson(
        "{\"detectAnomaly\": false, \"numStepsToPredict\": 20}"
      )) AS val,
      val.currentTimestamp AS currentOpTime,
      val.actualValue AS actual,
      val.predictedValue AS predict,
      val.adjustedSymmetricPercentageError AS aspe,
      val.percentageErrorBound AS errorBound,
      val.isAnomaly AS isAnomaly,
      val.nextTimestamp AS nextOpTime,
      val.nextPredictedValue AS nextPredicted 
      FROM forecast_win;
    

      SELECT tableName, forecast(operationNum, operationTime, 60) AS val,
      val.currentTimestamp AS currentOpTime,
      val.actualValue AS actual,
      val.predictedValue AS predict,
      val.adjustedSymmetricPercentageError AS aspe,
      val.percentageErrorBound AS errorBound,
      val.isAnomaly AS isAnomaly,
      val.nextTimestamp AS nextOpTime,
      val.nextPredictedValue AS nextPredicted 
      FROM forecast_win
      GROUP BY tableName;
    

    t1 = 14:00 → v1
    t2 = 14:01 → v2
    t3 = 14:03 → v3
  

    [t0 + (k-1) * s, t0 + k * s]
  

This example TQL application implements a real-time forecasting and anomaly detection pipeline for incident report data streamed from JSON log files. The application reads incident data, aggregates it in 3-minute windows, and applies a forecasting model to detect anomalies in incident rates. Alerts for anomalous activity are published to Slack, and forecasting results are stored in a WActionStore.

CREATE OR REPLACE APPLICATION Example_case1;

CREATE OR REPLACE SOURCE IncidentReports USING Global.FileReader ( 
  directory: '/Users/user/Documents/CSVData/incident_files_generated/', 
  adapterName: 'FileReader', 
  rolloverstyle: 'Default', 
  wildcard: '*.log', 
  blocksize: 64, 
  skipbom: true, 
  includesubdirectories: false, 
  positionbyeof: false ) 
PARSE USING Global.JSONParser ( 
  fieldName: '', 
  handler: 'com.webaction.proc.JSONParser_1_0', 
  parserName: 'JSONParser' ) 
OUTPUT TO IncidentReports_stream;

CREATE OR REPLACE TYPE IncidentWStype (
 currentOpTime org.joda.time.DateTime,
 actual java.lang.Double,
 predicted java.lang.Double,
 isAnomaly java.lang.Boolean);

CREATE OR REPLACE TYPE IncidentOutput_type (
 val com.webaction.analytics.forecast.ForecastResult,
 currentOpTime org.joda.time.DateTime,
 actual java.lang.Double,
 predicted java.lang.Double,
 aspe java.lang.Double,
 errorBound java.lang.Double,
 isAnomaly java.lang.Boolean,
 nextOpTime org.joda.time.DateTime,
 nextPredicted java.lang.Double);

CREATE OR REPLACE CQ parseInput 
INSERT INTO IncidentReports 
SELECT TO_DATE(TO_STRING(TO_DATE(a.data.get('incident_data').get('incident_time').textValue()), "yyyy-MM-dd HH:mm:00")) as incident_time, a.data.get('incident_data').get('cause') as cause
FROM IncidentReports_stream a;

CREATE OR REPLACE WACTIONSTORE IncidentWS CONTEXT OF IncidentWStype EVENT TYPES ( IncidentWStype KEY(currentOpTime) );

CREATE OR REPLACE STREAM IncidentForecastResult OF IncidentOutput_type;

CREATE OR REPLACE JUMPING WINDOW IncidentWindow OVER IncidentReports 
KEEP WITHIN 3 MINUTE ON incident_time;

CREATE OR REPLACE STREAM IncidentAlertStream of Global.AlertEvent;


CREATE OR REPLACE CQ IncidentAgg
INSERT INTO incidentWindow_Stream
SELECT count(*) as count, first(incident_time) as incident_time
FROM IncidentWindow;;

CREATE OR REPLACE CQ Incident_Analyser 
INSERT INTO IncidentForecastResult 
SELECT forecast(to_double(e.count),e.incident_time, 60, makeJson("{\"anomalyEpsilon\": 0.20}")) as val, 
val.currentTimestamp as currentOpTime,
val.actualValue as actual, 
val.predictedValue as predicted,
val.adjustedSymmetricPercentageError as aspe,
val.percentageErrorBound as errorBound,
val.isAnomaly as isAnomaly,
val.nextTimestamp as nextOpTime,
val.nextPredictedValue as nextPredicted
FROM incidentWindow_Stream e;

CREATE OR REPLACE CQ incidentAnomalyAlertGenerator
INSERT INTO IncidentAlertStream
SELECT "Incidents", "Actual: " + TO_STRING(e.actual) + "; Predicted: " + TO_STRING(e.predicted), "error", "notify", "We are seeing abnormal Incident rate."
FROM IncidentForecastResult e;;

CREATE OR REPLACE TARGET SlackTarget USING Global.SlackAlertAdapter ( 
  adapterName: 'SlackAlertAdapter', 
  OauthToken: 'OauthToken', 
  refreshtokenrenewal: false, 
  OauthToken_encrypted: 'true', 
  ChannelName: 'slack-test' ) 
INPUT FROM IncidentAlertStream;

CREATE OR REPLACE CQ IncidentWSWriter 
INSERT INTO IncidentWS 
SELECT currentOpTime, actual, predicted, isAnomaly
FROM IncidentForecastResult;
 
END APPLICATION Example_case1;