Embedding空間中的時序異常檢測，你學(xué)會了嗎？

作者：百度Geek說 2024-08-08 15:58:35

通過實驗，驗證了該算法的有效性，但在后續(xù)的工程化應(yīng)用中，還需要結(jié)合具體的應(yīng)用場景進行適當?shù)恼{(diào)整。比如采樣點的數(shù)量、采樣點的選取方法、樣本Embedding方法、距離計算方法等。

1.背景

在安全、反作弊等業(yè)務(wù)場景下，對流量、用戶行為進行異常檢測是基本的剛需。通常的做法是，在各個業(yè)務(wù)維度上，對流量、用戶行為進行統(tǒng)計分析，提取出相應(yīng)的指標特征，然后在時間維度上，對這些指標特征進行建模分析。再利用相關(guān)的算法來檢測當前的指標值是否背離了該指標在歷史數(shù)據(jù)中的分布規(guī)律。

2.示例

假設(shè)某業(yè)務(wù)場景下，用戶有100個來源渠道，用戶使用產(chǎn)品時，有10種不同的操作方式，對于用戶的行為，我們可以簡單的撮取出PV、UV、失敗率等指標。那么我們可以建立這樣一個監(jiān)控：

監(jiān)控的維度：來源渠道 * 操作方式 = 100 * 10 = 1000個維度

監(jiān)控的指標：PV、UV、失敗率...

統(tǒng)計周期: 小時

然后針對每個維度、時刻、指標，收集過去30天的數(shù)據(jù)做為訓(xùn)練樣本，訓(xùn)練異常檢測模型（如EllipticEnvelope等），然后對當前時刻的指標值，進行異常檢測。

上面的方法，通過合理的拆分監(jiān)控維度，一方面可以有效的提高檢測的靈敏度，避免較少的異常流量淹沒在大盤監(jiān)控在隨機波動中；另一方面，也可以對異常流量進行快速的定位，便于及時處理。

3.問題

上面的方法也存在諸多的限制，比如：

監(jiān)控維度必需是離散、可枚舉的，否則無法建立歷史數(shù)據(jù)的統(tǒng)計模型；
監(jiān)控維度的粒度必須合適，否則或是靈敏度不足，或是噪聲太多，無法有效檢測異常。

顯然，不是所有的業(yè)務(wù)場景都能滿足上述的要求。即便是能滿足上述要求的業(yè)務(wù)場景中，隨著對攻擊者的對抗不斷深入，攻擊者會嘗試降低攻擊的規(guī)模，并盡量將攻擊行為分散到更多的維度中，從而躲避我們的檢測手段。

4.解決思路

那么，能否不依賴業(yè)務(wù)維度拆分，直接對指標進行異常檢測呢？

首先，我們需要把待檢測的每一條日志、數(shù)據(jù)當做一個獨立的樣本。接下來，不難聯(lián)想到，這些樣本都可以映射到某個高維空間中，我們把這個空間叫做樣本空間。可以通過向量化、Embedding等方法，得到樣本在這個空間中的坐標。

樣本在這個空間中的分布必然不是完全隨機的，而是會存在一定的特點（分布特征）。若當前時刻樣本在這個空間中的分布特征與歷史數(shù)據(jù)中的分布特征不一致，則說明當前樣本存在異常。而分布在差異最大的區(qū)域中的樣本，則可以認為是異常樣本。

接下來的問題就變成了如何對這種分布特征進行建模？

最先想到的是，我們可以通過聚類算法，來對樣本進行劃分，再對每個Cluster，提取出統(tǒng)計特征。但在具體實現(xiàn)時還需要考慮以下問題：

支持的樣本數(shù)量要足夠多；
支持的Cluster數(shù)量要足夠多；
每個Cluster的樣本數(shù)量要盡可能均勻；
Cluster的劃分要盡可能穩(wěn)定，才能在時間維度上執(zhí)行異常檢測。

再進一步，其實我們不需要執(zhí)行完整的聚類算法，我們只需要對樣本空間設(shè)置足夠多的采樣點進行采樣，計算出采樣點附近的樣本的統(tǒng)計特征做為采集采樣點的分布特征，再對采樣點的特征進行時間維度的異常檢測，即可完成對整個樣本空間的異常檢測了。

圖片

5.算法實驗

5.1 數(shù)據(jù)準備

取某業(yè)務(wù)場景下近30天的用戶行為日志，約160萬條，利用其中的UserAgent信息，對其進行向量化處理。每條日志的向量長度為128維。

向量化算法：

def to_vector(ua):
    if isinstance(ua, (list, tuple)):
        return [to_vector(c) for c in ua]
    else:
        vec = np.zeros(128)
        for c in ua:
            vec[ord(c) % 128] += 1  # UserAgent中的字符絕大多數(shù)都是Ascll字符，所以取余128
        l2 = np.sqrt(np.sum(vec * vec))
        if l2 != 0:
            vec /= l2
        return vec.tolist()

將清洗好的數(shù)據(jù)保存到向量DB中備用：

for day in days:
    for hour in hours:
        event_day = day.strftime("%Y%m%d")
        event_hour = "{:02d}".format(hour)
        collection = chroma_client.get_or_create_collection(
            name="{}_{}_{}".format(name_prefix, event_day, event_hour)
        )
        sub_df = df_ua_pv[(df_ua_pv.event_day == event_day) & (df_ua_pv.event_hour == event_hour)]
        ids = [hashlib.md5(bytes(str(row), "utf-8")).hexdigest() for _, row in sub_df.iterrows()]
        docs = [row.ua for _, row in sub_df.iterrows()]
        metadatas = [{"pv": row.pv} for _, row in sub_df.iterrows()]
        embeddings = [to_vector(row.ua) for _, row in sub_df.iterrows()]
        batch_size = 10000
        for batch_id in range(0, len(docs), batch_size):
            collection.upsert(
                ids=ids[batch_id : batch_id + batch_size],
                documents=docs[batch_id : batch_id + batch_size],
                metadatas=metadatas[batch_id : batch_id + batch_size],
                embeddings=embeddings[batch_id : batch_id + batch_size],
            )
            print("{:>8d} / {}".format(batch_id + batch_size, len(docs)))
        collections[event_day + event_hour] = collection

為了更方便的驗證算法的有效性，在數(shù)據(jù)集中，人工構(gòu)造了一些異常樣本，包括：

個別隨機UA，PV增長：10%， 20%， 50%， 100%， 200%， 500%，1000%；數(shù)量：5；min_pv=100。
部分相似UA，PV增長：5%，10%，20%， 50%， 100%；數(shù)量：10， 20， 50， 100；min_pv=10。
生成相似UA，PV同比增長，數(shù)量：10， 20， 50， 100。
生成相似UA，整體PV不增長，數(shù)量：10， 20， 50， 100；min_pv=1。

5.2 算法實現(xiàn)

隨機生成采樣點：

query_ua_list = (
    df_ua_pv[(df_ua_pv.event_day == event_day) & (df_ua_pv.event_hour == event_hour)].sample(100)["ua"].to_list()
)

在樣本空間進行鄰近采樣：

results = []
query_ua_vec = to_vector(query_ua_list)
for day in days:
    for hour in hours:
        res = get_collection(day, hour).query(query_embeddings=query_ua_vec, n_results=n_results)
        for i in range(len(query_ua_list)):
            for j in range(n_results):
                row = [
                    query_ua_list[i],
                    res["metadatas"][i][j]["event_day"],
                    res["metadatas"][i][j]["event_hour"],
                    res["documents"][i][j],
                    res["metadatas"][i][j]["pv"],
                    res["distances"][i][j],
                ]
                if extra_fields:
                    for field in extra_fields:
                        row.append(res["metadatas"][i][j].get(field))
                results.append(row)
cols = ["ua", "day", "hour", "doc", "pv", "dist"]
if extra_fields:
    cols += extra_fields
df_results = pd.DataFrame(results, columns=cols)

定義要檢測的字段：

AREA_EXP = [0, 2, 8]
MODEL_FIELDS = ["pv", "dist"]
MODEL_FIELDS += [f"dens_{i}" for i in AREA_EXP]
MODEL_FIELDS += ["dens_s"]
MODEL_AGGS = {}
for col in MODEL_FIELDS:
    MODEL_AGGS[f"{col}_mean"] = (col, "mean")
    MODEL_AGGS[f"{col}_std"] = (col, "std")

進行天維度的異常檢測：

df_query_results["dens_s"] = 1 / (df_query_results["dist"] ** 0.5 + 1)
df_res_agg = df_query_results.groupby(["ua", "day"], as_index=False).agg(
    pv=("pv", "sum"),
    dist=("dist", "mean"),
    dens_s=("dens_s", "mean"),
)
for i in AREA_EXP:
    df_res_agg["area_{}".format(i)] = (df_res_agg["dist"] * 10) ** i
    df_res_agg["dens_{}".format(i)] = df_res_agg["pv"] / df_res_agg["area_{}".format(i)]
df_model = df_res_agg[df_res_agg.day <= last_event_day].groupby("ua").agg(**MODEL_AGGS)
df_check = df_res_agg.join(df_model, notallow="ua")
for col in MODEL_FIELDS:
    df_check[f"{col}_sigma"] = (df_check[col] - df_check[f"{col}_mean"]) / df_check[f"{col}_std"]
df_check["dens_avg_sigma"] = df_check[["dens_s_sigma"] + [f"dens_{i}_sigma" for i in AREA_EXP]].mean(axis=1)
df_check["dens_max_sigma"] = df_check[["dens_s_sigma"] + [f"dens_{i}_sigma" for i in AREA_EXP]].max(axis=1)
df_check["dens_min_sigma"] = df_check[["dens_s_sigma"] + [f"dens_{i}_sigma" for i in AREA_EXP]].min(axis=1)