人工智能个人理财助手的用户行为分析：从埋点到洞察的完整实战

// track.js
import { Analytics } from 'mobile-analytics-sdk';

export const trackBillAdd = (bill) => {
  Analytics.track('bill_add', {
    bill_type: bill.type,
    amount: bill.amount,
    auto_tag: bill.autoTag,
    ts: Date.now(),
    user_id: global.userId,
  });
};

# logger_service.py
from fastapi import FastAPI, Request
from aiokafka import AIOKafkaProducer
import json, time, uvicorn

app = FastAPI()
producer = AIOKafkaProducer(bootstrap_servers='kafka:9092')

@app.post("/collect")
async def collect(request: Request):
    payload = await request.json()
    payload['server_ts'] = int(time.time()*1000)
    await producer.send("user_events", json.dumps(payload).encode())
    return {"status": "ok"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

# etl_stream.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder.appName("IPA_ETL").getOrCreate()

df = (spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("subscribe", "user_events")
        .load()
        .selectExpr("CAST(value AS STRING) as json"))

schema = spark.read.json(df.rdd.map(lambda r: r.json)).schema

parsed = (df
  .select(from_json(col("json"), schema).alias("e"))
  .select("e.*")
  .withColumn("dt", to_date(from_unixtime(col("ts")/1000)))
  .withWatermark("ts", "10 minutes"))

query = (parsed.writeStream
           .format("parquet")
           .option("path", "/warehouse/ods/events")
           .option("checkpointLocation", "/chk/events")
           .partitionBy("dt")
           .trigger(processingTime='1 minute')
           .start())
query.awaitTermination()

# user_tag.py
from pyspark.sql import functions as F
from pyspark.ml.feature import Bucketizer

df = spark.table("dws.user_daily")

# 计算「理财成熟度」
df = df.withColumn("maturity_score",
    F.when(F.col("invest_view_cnt_30d") > 5, 3)
     .when(F.col("budget_set_cnt_7d") > 0, 2)
     .otherwise(1))

# 金额分桶
bucketizer = Bucketizer(splits=[0, 500, 2000, 10000, float("inf")],
                        inputCol="avg_bill_amount_30d",
                        outputCol="amount_level")
df = bucketizer.transform(df)

df.select("user_id", "maturity_score", "amount_level") \
  .write.mode("overwrite").saveAsTable("ads.user_tag")

-- retention.sql
WITH first_day AS (
  SELECT user_id, min(dt) AS first_dt
  FROM ods.events
  WHERE event='app_launch'
  GROUP BY 1
),
retention AS (
  SELECT f.user_id,
         datediff(e.dt, f.first_dt) AS delta
  FROM first_day f
  JOIN ods.events e
    ON f.user_id = e.user_id
)
SELECT delta, count(distinct user_id) as users
FROM retention
WHERE delta BETWEEN 0 AND 30
GROUP BY delta
ORDER BY delta;

import pandas as pd, seaborn as sns, matplotlib.pyplot as plt
df = pd.read_sql("SELECT * FROM retention ORDER BY delta", conn)
sns.lineplot(x='delta', y='users', data=df)
plt.title("Cohort Retention")
plt.show()

import retentioneering as re
re.config.update_config(col_event='event',
                        col_user_id='user_id',
                        col_time='ts')
df = pd.read_parquet("/warehouse/ods/events")
dataset = re.Dataset(df)
dataset.rete.plot_graph(
    weight_col='user_id',
    thresh=0.03,
    targets=['invest_view'],
    width=800, height=600
)

from lifetimes import GammaGammaFitter, BetaGeoFitter
df_rfm = spark.sql("""
  SELECT user_id,
         datediff(max(dt), min(dt)) as T,
         count(distinct dt) as frequency,
         sum(amount) as monetary_value
  FROM ods.events
  WHERE event='bill_add'
  GROUP BY 1
""").toPandas()

bgf = BetaGeoFitter(penalizer_coef=0.01)
bgf.fit(df_rfm['frequency'], df_rfm['T'])

ggf = GammaGammaFitter(penalizer_coef=0.01)
ggf.fit(df_rfm['frequency'], df_rfm['monetary_value'])

df_rfm['predicted_clv'] = ggf.customer_lifetime_value(
    bgf,
    df_rfm['frequency'],
    df_rfm['T'],
    df_rfm['monetary_value'],
    time=12,  # 12 个月
    freq_unit='D'
)

import gym, numpy as np
from stable_baselines3 import PPO

class FinancePushEnv(gym.Env):
    def __init__(self, user_features):
        super().__init__()
        self.user = user_features
        self.action_space = gym.spaces.Discrete(4)
        self.observation_space = gym.spaces.Box(low=0, high=1, shape=(10,), dtype=np.float32)
    def step(self, action):
        reward = simulate_reward(self.user, action)
        next_state = self.user.next_features()
        done = True
        return next_state, reward, done, {}
    def reset(self):
        return self.user.to_vec()

env = FinancePushEnv(user_features)
model = PPO("MlpPolicy", env, verbose=1).learn(total_timesteps=10000)

# ab_test.py
import statsmodels.api as sm

def evaluate_exp(df):
    treat = df[df.variant=='push_v2']['converted']
    ctrl  = df[df.variant=='push_v1']['converted']
    return sm.stats.proportion_ztest([treat.sum(), ctrl.sum()],
                                     [len(treat), len(ctrl)])

p_value = evaluate_exp(spark.table("exp.push_exp").toPandas())
print("p-value:", p_value)