Как я разрабатывал ИИ‑сервис для проверки кошельков и транзакций (AML). Код проекта.

# model.py import numpy as np import pandas as pd from sklearn.linear_model import LogisticRegression from sklearn.model_selection import train_test_split from joblib import dump, load import os MODEL_PATH = "misttrack_model.joblib" FEATURE_COLUMNS = ["tx_count", "incoming_ratio", "avg_amount", "unique_counterparties", "recent_activity_days"] def generate_synthetic_dataset(n=5000, random_state=42): rng = np.random.RandomState(random_state) tx_count = rng.poisson(10, size=n) + 1 incoming_ratio = rng.beta(2, 5, size=n) # доля входящих avg_amount = np.exp(rng.normal(2, 1, size=n)) # лог-норм распределение unique_counterparties = np.clip((tx_count * rng.uniform(0.2, 1.0, size=n)).astype(int), 1, None) recent_activity_days = rng.exponential(30, size=n) # дни с последней активности # Простая, синтетическая «целевая» метка риска для обучения: # чем больше tx_count и avg_amount и чем ближе recent_activity - тем выше риск risk_score = ( 0.3 * (tx_count / (tx_count.max()+1)) + 0.4 * (np.log1p(avg_amount) / np.log1p(avg_amount).max()) + 0.2 * (1 - incoming_ratio) + # например: больше исходящих — показатель (в синтетике) 0.1 * (1 - np.tanh(recent_activity_days / 30)) ) # преобразуем в бинарную метку через порог y = (risk_score + 0.05 * rng.normal(size=n)) > 0.6 df = pd.DataFrame({ "tx_count": tx_count, "incoming_ratio": incoming_ratio, "avg_amount": avg_amount, "unique_counterparties": unique_counterparties, "recent_activity_days": recent_activity_days, "label": y.astype(int) }) return df def train_and_save(model_path=MODEL_PATH): df = generate_synthetic_dataset() X = df[FEATURE_COLUMNS] y = df["label"] X_train, X_test, y_train, y_test = train_test_split(X, y, stratify=y, test_size=0.2, random_state=1) model = LogisticRegression(max_iter=1000) model.fit(X_train, y_train) print("Train score:", model.score(X_train, y_train)) print("Test score:", model.score(X_test, y_test)) dump({"model": model, "features": FEATURE_COLUMNS}, model_path) print(f"Model saved to {model_path}") if __name__ == "__main__": if not os.path.exists(MODEL_PATH): train_and_save() else: print("Model already exists.")

# fetcher.py import random import datetime from typing import List, Dict # В реальной системе здесь был бы клиент к public node / block explorer API. # В демо мы возвращаем синтетические транзакции. def simulate_transactions_for_address(address: str, seed: int = None): rng = random.Random(seed or hash(address) & 0xffffffff) tx_count = rng.randint(1, 50) now = datetime.datetime.utcnow() txs = [] for i in range(tx_count): direction = rng.choice(["in", "out"]) amount = round(rng.uniform(0.0001, 50.0) * (1 if rng.random() < 0.8 else 10), 8) days_ago = rng.expovariate(1/30) timestamp = now - datetime.timedelta(days=days_ago) counterparty = "0x" + "".join(rng.choice("0123456789abcdef") for _ in range(40)) txs.append({ "txid": f"tx{i}_{address[:6]}", "direction": direction, "amount": amount, "timestamp": timestamp, "counterparty": counterparty }) # сортируем по времени txs = sorted(txs, key=lambda x: x["timestamp"], reverse=True) return txs

# app.py from flask import Flask, render_template, request from joblib import load from fetcher import simulate_transactions_for_address from features import extract_features_from_txs, FEATURE_COLUMNS import os import model as model_module MODEL_PATH = "misttrack_model.joblib" app = Flask(__name__) def ensure_model(): if not os.path.exists(MODEL_PATH): print("Training model (synthetic data)...") model_module.train_and_save(MODEL_PATH) data = load(MODEL_PATH) return data["model"], data["features"] model, feature_cols = ensure_model() @app.route("/", methods=["GET", "POST"]) def index(): if request.method == "POST": address = request.form.get("address", "").strip() if not address: return render_template("index.html", error="Введите адрес.") # В ДЕМO — получаем синтетические транзакции txs = simulate_transactions_for_address(address) feats = extract_features_from_txs(txs) X = [[feats[c] for c in feature_cols]] prob = float(model.predict_proba(X)[0][1]) pred = int(prob > 0.5) # объяснение — коэффициенты логистической регрессии coef = model.coef_[0] explanations = [] for name, c, val in zip(feature_cols, coef, X[0]): impact = c * val explanations.append({"feature": name, "coef": float(c), "value": float(val), "impact": float(impact)}) explanations = sorted(explanations, key=lambda x: abs(x["impact"]), reverse=True) report = { "address": address, "probability": prob, "flag": "Повышенный риск" if pred else "Низкий/средний риск", "features": feats, "explanations": explanations, "sample_txs": txs[:10] # показываем несколько транзакций } return render_template("report.html", report=report) return render_template("index.html") if __name__ == "__main__": app.run(debug=True, port=5000)