Detect unauthorized data transfers using timeseries anomaly (ASIM Web Session)

Original query, unchanged.

let startTime = 14d;
let endTime = 1d;
let timeframe = 1h;
let scorethreshold = 5;
let bytessentperhourthreshold = 10;
// calculate avg. eps(events per second)
let eps = materialize(_Im_WebSession(starttime=ago(1d))
    | project TimeGenerated
    | summarize AvgPerSec = count() / 3600 by bin(TimeGenerated, 1h)
    | summarize round(avg(AvgPerSec))
    );
let summarizationexist  = (
    union isfuzzy=true 
        (
        WebSessionEssentialsCustomParser
        | where Type in ("WebSession_Summarized_SrcIP_CL", "WebSession_Summarized_SrcIPV1_CL")
        | where EventTime_t > ago(1d) 
        | project v = int(2)
        ),
        (
        print int(1) 
        | project v = print_0
        )
    | summarize maxv = max(v)
    | extend sumexist = (maxv > 1)
    );
let TimeSeriesData = union isfuzzy=true 
        (
        (datatable(exists: int, sumexist: bool)[1, false]
        | where toscalar(eps) > 1000
        | join (summarizationexist) on sumexist)
        | join (
            _Im_WebSession(starttime=ago(2d), endtime=now())
            | project DstIpAddr, SrcBytes, TimeGenerated, EventProduct
            | where isnotempty(DstIpAddr)
                and not(ipv4_is_private(DstIpAddr))
                and isnotempty(SrcBytes)
            | summarize SrcBytesSum=tolong(sum(SrcBytes)) by EventProduct, bin(TimeGenerated, 1h)
            | extend EventTime = TimeGenerated, exists=int(1)
            )
            on exists
        | project-away exists*, maxv, sum*
        ),
        (
        (datatable(exists: int, sumexist: bool)[1, false]
        | where toscalar(eps) between (501 .. 1000)
        | join (summarizationexist) on sumexist)
        | join (
            _Im_WebSession(starttime=ago(3d), endtime=now())
            | project DstIpAddr, SrcBytes, TimeGenerated, EventProduct
            | where isnotempty(DstIpAddr)
                and not(ipv4_is_private(DstIpAddr))
                and isnotempty(SrcBytes)
            | summarize SrcBytesSum=tolong(sum(SrcBytes)) by EventProduct, bin(TimeGenerated, 1h)
            | extend EventTime = TimeGenerated, exists=int(1)
            )
            on exists
        | project-away exists*, maxv, sum*
        ),
        (
        (datatable(exists: int, sumexist: bool)[1, false]
        | where toscalar(eps) <= 500
        | join (summarizationexist) on sumexist)
        | join (
            _Im_WebSession(starttime=ago(4d), endtime=now())
            | project DstIpAddr, SrcBytes, TimeGenerated, EventProduct
            | where isnotempty(DstIpAddr)
                and not(ipv4_is_private(DstIpAddr))
                and isnotempty(SrcBytes)
            | summarize SrcBytesSum=tolong(sum(SrcBytes)) by EventProduct, bin(TimeGenerated, 1h)
            | extend EventTime = TimeGenerated, exists=int(1)
            )
            on exists
        | project-away exists*, maxv, sum*
        ),
        (
        WebSessionEssentialsCustomParser
        | where Type in ("WebSession_Summarized_SrcIP_CL", "WebSession_Summarized_SrcIPV1_CL")
        | where EventTime_t between (ago(startTime) .. now())
        | where isnotempty(SrcBytes_d) and not(DstIPIsPrivate_b)
        | project
            SrcBytesSum=tolong(SrcBytes_d),
            EventTime=EventTime_t,
            EventProduct = EventProduct_s
        )
    | make-series TotalBytesSent = sum(SrcBytesSum) on EventTime from startofday(ago(startTime)) to startofday(now()) step timeframe by EventProduct;
// TimeSeriesData block ends here
//Take only anomalies in TimeSeriesData
let TimeSeriesAnomalies = materialize(TimeSeriesData
    | extend (anomalies, score, baseline) = series_decompose_anomalies(TotalBytesSent, scorethreshold, -1, 'linefit')
    | mv-expand
        TotalBytesSent to typeof(long),
        EventTime to typeof(datetime),
        anomalies to typeof(double),
        score to typeof(double),
        baseline to typeof(long)
    | where anomalies > 0 and baseline > 0
    | extend AnomalyHour = EventTime
    | extend
        TotalBytesSentinMBperHour = round(((TotalBytesSent / 1024) / 1024), 2),
        BaselineBytesSentinMBperHour = round(((baseline / 1024) / 1024), 2),
        score = round(score, 2)
    | project
        EventProduct,
        AnomalyHour,
        TotalBytesSentinMBperHour,
        BaselineBytesSentinMBperHour,
        anomalies,
        score
    | where AnomalyHour between (startofday(ago(endTime)) .. startofday(now())) // Get TimeSeriesAnomalies in previous day
        );
// TimeSeriesAlerts block end here
let AnomalyHours = materialize (TimeSeriesAnomalies
    | project AnomalyHour);
//Previous day aggregated per hour
let PreviousDayLogs = 
    _Im_WebSession(starttime=startofday(ago(endTime)), endtime=startofday(now()))
    | where isnotempty(DstIpAddr) and isnotempty(SrcIpAddr) and isnotempty(SrcBytes)
    | where not(ipv4_is_private(DstIpAddr))
    | project
        TimeGenerated,
        DstIpAddr,
        SrcIpAddr,
        SrcBytes,
        DstBytes,
        DstPortNumber,
        EventProduct
    | extend DateHour = bin(TimeGenerated, timeframe) // create a new column and round to hour
    | where DateHour in (AnomalyHours) // Filter dataset to include only anomaly AnomalyHours
    | extend
        SentBytesinMB = ((SrcBytes / 1024) / 1024),
        ReceivedBytesinMB = ((DstBytes / 1024) / 1024)
    | summarize
        HourlyCount = count(),
        TimeGeneratedMax = arg_max(TimeGenerated, *),
        DestinationIPList = make_set(DstIpAddr, 100),
        DestinationPortList = make_set(DstPortNumber, 100),
        SentBytesinMB = tolong(sum(SentBytesinMB)),
        ReceivedBytesinMB = tolong(sum(ReceivedBytesinMB))
        by SrcIpAddr, EventProduct, TimeGeneratedHour = bin(TimeGenerated, timeframe)
    | where SentBytesinMB > bytessentperhourthreshold
    | sort by TimeGeneratedHour asc, SentBytesinMB desc
    | extend Rank=row_number(1, prev(TimeGeneratedHour) != TimeGeneratedHour) // Ranking the dataset per Hourly Partition
    | where Rank <= 10  // Selecting Top 10 records with Highest BytesSent in each Hour
    | project
        EventProduct,
        TimeGeneratedHour,
        TimeGeneratedMax,
        SrcIpAddr,
        DestinationIPList,
        DestinationPortList,
        SentBytesinMB,
        ReceivedBytesinMB,
        Rank,
        HourlyCount;
// PreviousDayLogs block ends here
TimeSeriesAnomalies
| join kind = inner (PreviousDayLogs
    | extend AnomalyHour = TimeGeneratedHour)
    on EventProduct, AnomalyHour
| sort by score desc
| project
    EventProduct,
    AnomalyHour,
    TimeGeneratedMax,
    SrcIpAddr,
    DestinationIPList,
    DestinationPortList,
    SentBytesinMB,
    ReceivedBytesinMB,
    TotalBytesSentinMBperHour,
    BaselineBytesSentinMBperHour,
    score,
    anomalies,
    HourlyCount
| summarize
    EventCount = sum(HourlyCount),
    startTimeUtc = min(TimeGeneratedMax),
    EndTimeUtc = max(TimeGeneratedMax),
    SentBytesinMB = sum(SentBytesinMB),
    ReceivedBytesinMB = sum(ReceivedBytesinMB),
    SourceIP = take_any(SrcIpAddr),
    SourceIPList = make_set(SrcIpAddr, 10),
    DestinationIPList = make_set(DestinationIPList, 100),
    DestinationPortList = make_set(DestinationPortList, 100)
    by
    AnomalyHour,
    TotalBytesSentinMBperHour,
    BaselineBytesSentinMBperHour,
    score,
    anomalies,
    EventProduct
| project
    EventProduct,
    AnomalyHour,
    startTimeUtc,
    EndTimeUtc,
    SourceIP,
    SourceIPList,
    DestinationIPList,
    DestinationPortList,
    SentBytesinMB,
    ReceivedBytesinMB,
    TotalBytesSentinMBperHour,
    BaselineBytesSentinMBperHour,
    anomalies,
    score,
    EventCount