4、特征工程实战：时间序列特征提取、用户行为画像构建、IP与地理位置特征

特征工程，说白了就是把原始日志变成机器能理解的语言。我做了这么多年安全运维，见过太多好模型被烂特征拖垮的例子。你想想看，原始登录日志里无非是时间、用户、IP、结果这些字段，但怎么把它们变成能反映异常的信号，这才是真功夫。

这一章，我们就来拆解这个过程。我会带着你，一步步把原始日志加工成特征矩阵。

核心思路：特征工程不是堆砌字段，而是把安全经验编码成数值。好的特征，能让简单的模型也表现惊艳。

4.1 时间序列特征提取

时间，是审计日志里最容易被忽视的金矿。我刚开始做这块时，只盯着登录结果看，后来才发现，时间模式本身就能暴露大量问题。

4.1.1 基础时间特征

从时间戳里，我们能直接提取出这些特征：

  • 小时（Hour）：凌晨3点的登录，比下午3点可疑得多
  • 星期几（Day of Week）：周末的批量登录，可能是攻击脚本
  • 是否工作日（Is Workday）：法定节假日出现的登录，需要关注
  • 时段分类（Time Period）：分为凌晨、上午、下午、晚间
import pandas as pd

# 假设 df 包含 'timestamp' 列
df['timestamp'] = pd.to_datetime(df['timestamp'])
df['hour'] = df['timestamp'].dt.hour
df['day_of_week'] = df['timestamp'].dt.dayofweek  # 0=周一, 6=周日
df['is_workday'] = df['day_of_week'].apply(lambda x: 1 if x < 5 else 0)

# 时段分类
def classify_period(hour):
    if 0 <= hour < 6:
        return 0  # 凌晨
    elif 6 <= hour < 12:
        return 1  # 上午
    elif 12 <= hour < 18:
        return 2  # 下午
    else:
        return 3  # 晚间

df['time_period'] = df['hour'].apply(classify_period)

4.1.2 滑动窗口统计特征

单条日志的时间点信息有限，真正的价值在于「一段时间内的行为模式」。我个人习惯用滑动窗口来提取这些特征：

  • 过去1小时内登录次数：高频登录可能是暴力破解
  • 过去1小时内失败次数：连续失败是典型攻击信号
  • 过去24小时内不同IP数：账号被多地登录，可能已泄露
  • 两次登录间隔：间隔极短，可能是自动化脚本
# 按用户分组，计算滑动窗口特征
df = df.sort_values(['user', 'timestamp'])

# 过去1小时内登录次数
df['login_count_1h'] = df.groupby('user')['timestamp'].transform(
    lambda x: x.rolling(window='1h', min_periods=1).count()
)

# 过去1小时内失败次数
df['fail_count_1h'] = df.groupby('user')['timestamp'].transform(
    lambda x: x.rolling(window='1h', min_periods=1).apply(
        lambda y: (df.loc[y.index, 'status'] == 'FAIL').sum()
    )
)

# 两次登录间隔（秒）
df['time_since_last'] = df.groupby('user')['timestamp'].diff().dt.total_seconds()

避坑指南：我曾经在滑动窗口上吃过亏——窗口大小选错了。对于内部员工，1小时窗口可能太长；对于外部攻击，可能又太短。我建议你先做探索性分析，看看正常行为和异常行为的时间分布差异，再定窗口大小。

4.2 用户行为画像构建

每个用户都有自己的「行为指纹」。比如，小王每天9点上班，登录公司VPN；小李是运维，凌晨也会登录服务器。这些模式一旦被打破，就值得警惕。

4.2.1 历史行为基线

构建用户画像的第一步，是建立历史行为基线：

  • 常用登录时段：用户通常在什么时间登录
  • 常用登录IP：用户通常从哪些IP登录
  • 常用登录设备：用户通常用什么设备
  • 登录频率：用户每天平均登录几次
# 构建用户画像
user_profile = df.groupby('user').agg({
    'hour': lambda x: x.mode().iloc[0] if not x.mode().empty else -1,  # 常用时段
    'ip': lambda x: x.value_counts().index.tolist()[:3],  # 常用IP（前3个）
    'timestamp': 'count'  # 总登录次数
}).rename(columns={'timestamp': 'total_logins'})

# 计算用户平均每日登录次数
user_profile['avg_daily_logins'] = df.groupby('user')['timestamp'].apply(
    lambda x: x.dt.date.nunique()
) / user_profile['total_logins']

4.2.2 行为偏离度特征

有了基线，我们就可以计算当前行为与基线的偏离程度：

  • 时段偏离度：当前登录时段是否与常用时段一致
  • IP偏离度：当前IP是否在常用IP列表中
  • 频率偏离度：当前登录频率是否远超历史均值
# 标记行为偏离
df['is_unusual_hour'] = df.apply(
    lambda row: 1 if row['hour'] != user_profile.loc[row['user'], 'usual_hour'] else 0,
    axis=1
)

df['is_unusual_ip'] = df.apply(
    lambda row: 0 if row['ip'] in user_profile.loc[row['user'], 'usual_ips'] else 1,
    axis=1
)

# 频率偏离：当前1小时内登录次数是否超过历史均值的3倍
df['is_high_frequency'] = df['login_count_1h'] > (user_profile['avg_daily_logins'] * 3)

注意：用户画像需要定期更新。我见过一个案例，某员工换了部门，作息全变了，但画像还是旧的，结果天天报警。建议每周或每月重新计算一次基线。

4.3 IP与地理位置特征

IP地址里藏着大量信息。一个来自国外的登录请求，如果用户平时只在国内活动，那基本可以判定为异常。

4.3.1 IP基础特征

  • IP类型：公网IP、内网IP、CDN IP、代理IP
  • IP段：/24 或 /16 子网
  • ASN（自治系统号）：标识IP所属的网络运营商
  • 是否已知恶意IP：与威胁情报库比对
import ipaddress

# 判断IP类型
def classify_ip(ip_str):
    try:
        ip = ipaddress.ip_address(ip_str)
        if ip.is_private:
            return 'private'
        elif ip.is_loopback:
            return 'loopback'
        else:
            return 'public'
    except:
        return 'invalid'

df['ip_type'] = df['ip'].apply(classify_ip)

# 提取IP段（/24）
df['ip_subnet'] = df['ip'].apply(
    lambda x: str(ipaddress.ip_network(f'{x}/24', strict=False))
)

4.3.2 地理位置特征

地理位置特征需要借助IP地理库（如GeoIP）。这里我们假设已经有一个geo_df包含IP到地理位置的映射：

  • 国家/地区：登录来源国家
  • 城市：登录来源城市
  • 经纬度：可用于计算登录距离
  • 时区：与用户常用时区对比
# 假设 geo_df 包含 'ip', 'country', 'city', 'latitude', 'longitude'
df = df.merge(geo_df, on='ip', how='left')

# 计算登录距离（与用户常用位置的距离）
from math import radians, sin, cos, sqrt, atan2

def haversine(lat1, lon1, lat2, lon2):
    R = 6371  # 地球半径（公里）
    dlat = radians(lat2 - lat1)
    dlon = radians(lon2 - lon1)
    a = sin(dlat/2)**2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon/2)**2
    c = 2 * atan2(sqrt(a), sqrt(1-a))
    return R * c

# 计算与用户常用位置的距离
df['distance_km'] = df.apply(
    lambda row: haversine(
        user_profile.loc[row['user'], 'usual_lat'],
        user_profile.loc[row['user'], 'usual_lon'],
        row['latitude'],
        row['longitude']
    ),
    axis=1
)

经验之谈：地理位置特征虽然强大，但要注意隐私合规。我在金融行业做项目时，客户明确要求不能存储精确经纬度，只能用到城市级别。另外，VPN和代理会干扰地理位置判断，建议结合IP类型特征一起使用。

4.4 实战：用Pandas构建登录日志特征矩阵

好了，理论说完了，我们来动手。下面是一个完整的特征矩阵构建流程，我会把前面讲的所有特征整合到一起。

4.4.1 数据准备

假设我们有一份登录日志，包含以下字段：

字段名 说明 示例
timestamp 登录时间 2024-01-15 03:22:10
user 用户名 zhangsan
ip 登录IP 203.0.113.42
status 登录结果 SUCCESS / FAIL
device 设备指纹 Mozilla/5.0 ...

4.4.2 特征矩阵构建

def build_feature_matrix(df):
    """
    构建完整的特征矩阵
    """
    # 1. 时间特征
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df['hour'] = df['timestamp'].dt.hour
    df['day_of_week'] = df['timestamp'].dt.dayofweek
    df['is_workday'] = (df['day_of_week'] < 5).astype(int)
    df['time_period'] = df['hour'].apply(classify_period)
    
    # 2. 滑动窗口特征
    df = df.sort_values(['user', 'timestamp'])
    df['login_count_1h'] = df.groupby('user')['timestamp'].transform(
        lambda x: x.rolling(window='1h', min_periods=1).count()
    )
    df['fail_count_1h'] = df.groupby('user')['timestamp'].transform(
        lambda x: x.rolling(window='1h', min_periods=1).apply(
            lambda y: (df.loc[y.index, 'status'] == 'FAIL').sum()
        )
    )
    df['time_since_last'] = df.groupby('user')['timestamp'].diff().dt.total_seconds()
    
    # 3. IP特征
    df['ip_type'] = df['ip'].apply(classify_ip)
    df['ip_subnet'] = df['ip'].apply(
        lambda x: str(ipaddress.ip_network(f'{x}/24', strict=False))
    )
    
    # 4. 用户画像特征（假设已有 user_profile）
    df['is_unusual_hour'] = df.apply(
        lambda row: 1 if row['hour'] != user_profile.loc[row['user'], 'usual_hour'] else 0,
        axis=1
    )
    df['is_unusual_ip'] = df.apply(
        lambda row: 0 if row['ip'] in user_profile.loc[row['user'], 'usual_ips'] else 1,
        axis=1
    )
    
    # 5. 选择最终特征列
    feature_cols = [
        'hour', 'day_of_week', 'is_workday', 'time_period',
        'login_count_1h', 'fail_count_1h', 'time_since_last',
        'ip_type', 'is_unusual_hour', 'is_unusual_ip'
    ]
    
    # 对类别特征进行独热编码
    df_encoded = pd.get_dummies(df[feature_cols], columns=['ip_type', 'time_period'])
    
    return df_encoded

# 执行构建
feature_matrix = build_feature_matrix(df)
print(f"特征矩阵形状: {feature_matrix.shape}")
print(f"特征列: {feature_matrix.columns.tolist()}")

4.4.3 特征矩阵可视化

构建完成后，我习惯先看一眼特征分布，确认没有明显的问题：

# 检查缺失值
print("缺失值统计:")
print(feature_matrix.isnull().sum())

# 检查数值范围
print("\n数值统计:")
print(feature_matrix.describe())

# 检查类别平衡
print("\n类别分布:")
print(feature_matrix['is_unusual_ip'].value_counts())

常见坑：时间序列特征容易引入未来信息。比如用「过去1小时失败次数」时，如果窗口计算包含了当前时间之后的数据，那就泄露了。我建议在构建特征时，始终用当前时间点之前的数据来计算，不要用未来的数据。

4.5 本章知识体系

下面这张图，帮你理清特征工程的完整脉络：

特征工程知识体系 原始登录日志 时间序列特征 用户行为画像 IP与地理位置 基础时间特征 滑动窗口统计 历史行为基线 行为偏离度 IP基础特征 地理位置特征 特征矩阵（Feature Matrix） 特征工程核心原则 1. 每个特征都要有明确的安全含义，不要盲目堆砌 2. 时间窗口大小需要根据业务场景调整，没有万能参数 3. 用户画像需要定期更新，避免概念漂移

特征矩阵构建完成后，下一步就是喂给模型了。不过在那之前，我建议你先做一轮特征筛选，去掉那些相关性过高或者方差太小的特征。嗯，这部分内容，我们后面会详细讲。

← 上一章 📖 返回目录 下一章 →