Xây dựng hệ thống cảnh báo giao dịch với AWS Lambda
2024-03-21 — QuantTrade
Giới thiệu
Trong thế giới giao dịch tài chính hiện đại, tốc độ phản ứng với biến động thị trường là yếu tố quyết định thành công. Hệ thống cảnh báo giao dịch tự động giúp nhà đầu tư theo dõi thị trường 24/7 mà không cần phải liên tục theo dõi màn hình. AWS Lambda, với khả năng chạy mã theo sự kiện và khả năng mở rộng tự động, là một công cụ lý tưởng để xây dựng hệ thống cảnh báo như vậy.
Bài viết này sẽ hướng dẫn bạn cách xây dựng một hệ thống cảnh báo giao dịch sử dụng AWS Lambda, kết hợp với các dịch vụ AWS khác như API Gateway, DynamoDB và SNS.
Kiến trúc hệ thống
Hệ thống cảnh báo giao dịch của chúng ta sẽ có các thành phần sau:
- Nguồn dữ liệu thị trường: API từ các sàn giao dịch hoặc nhà cung cấp dữ liệu tài chính.
- API Gateway: Cổng vào cho dữ liệu đến từ nguồn bên ngoài.
- AWS Lambda: Xử lý logic kiểm tra điều kiện và kích hoạt cảnh báo.
- DynamoDB: Lưu trữ cấu hình cảnh báo và lịch sử.
- Amazon SNS (Simple Notification Service): Gửi thông báo qua nhiều kênh.
- Kênh thông báo: Email, SMS, ứng dụng di động.
Bước 1: Thiết lập Tài khoản AWS và IAM
Trước khi bắt đầu, bạn cần:
- Tạo tài khoản AWS nếu chưa có: https://aws.amazon.com
- Tạo người dùng IAM với quyền phù hợp:
- AWSLambdaFullAccess
- AmazonAPIGatewayAdministrator
- AmazonDynamoDBFullAccess
- AmazonSNSFullAccess
# Cài đặt AWS CLI và cấu hình
# pip install awscli
# aws configure
Bước 2: Tạo Bảng DynamoDB để Lưu Trữ Cấu Hình Cảnh Báo
import boto3
# Khởi tạo DynamoDB client
dynamodb = boto3.resource('dynamodb')
# Tạo bảng cấu hình cảnh báo
table = dynamodb.create_table(
TableName='TradingAlerts',
KeySchema=[
{
'AttributeName': 'alert_id',
'KeyType': 'HASH' # Partition key
}
],
AttributeDefinitions=[
{
'AttributeName': 'alert_id',
'AttributeType': 'S'
}
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5
}
)
# Chờ cho đến khi bảng được tạo
table.meta.client.get_waiter('table_exists').wait(TableName='TradingAlerts')
print("Bảng TradingAlerts đã được tạo thành công!")
# Tạo ví dụ về một cảnh báo
alert = {
'alert_id': 'alert_001',
'user_id': 'user_123',
'symbol': 'BTCUSDT',
'condition': 'price_above',
'threshold': 50000.0,
'active': True,
'notification_channels': ['email', 'sms'],
'contact_info': {
'email': 'user@example.com',
'phone': '+84123456789'
}
}
# Thêm cảnh báo vào bảng
table.put_item(Item=alert)
print("Đã thêm cảnh báo mẫu vào bảng")
Bước 3: Tạo Topic SNS cho Thông Báo
import boto3
# Khởi tạo SNS client
sns = boto3.client('sns')
# Tạo SNS topic
response = sns.create_topic(Name='TradingAlertNotifications')
topic_arn = response['TopicArn']
print(f"SNS Topic ARN: {topic_arn}")
# Đăng ký email nhận thông báo
email = 'your-email@example.com'
sns.subscribe(
TopicArn=topic_arn,
Protocol='email',
Endpoint=email
)
print(f"Đã đăng ký {email} để nhận thông báo")
# Đăng ký số điện thoại nhận SMS
phone_number = '+84123456789' # Thay bằng số điện thoại thực
sns.subscribe(
TopicArn=topic_arn,
Protocol='sms',
Endpoint=phone_number
)
print(f"Đã đăng ký {phone_number} để nhận SMS")
Bước 4: Tạo AWS Lambda Function
Bây giờ, chúng ta sẽ tạo một hàm Lambda để kiểm tra điều kiện giao dịch và gửi cảnh báo:
import json
import boto3
import os
import requests
from decimal import Decimal
# Khởi tạo clients
dynamodb = boto3.resource('dynamodb')
sns = boto3.client('sns')
# Cấu hình
TRADING_ALERTS_TABLE = 'TradingAlerts'
SNS_TOPIC_ARN = os.environ['SNS_TOPIC_ARN'] # Sẽ được cấu hình trong biến môi trường Lambda
BINANCE_API_URL = 'https://api.binance.com/api/v3/ticker/price'
def lambda_handler(event, context):
try:
# Lấy tất cả cảnh báo đang hoạt động
table = dynamodb.Table(TRADING_ALERTS_TABLE)
response = table.scan(
FilterExpression='active = :active',
ExpressionAttributeValues={':active': True}
)
active_alerts = response['Items']
# Lấy danh sách các symbol cần kiểm tra
symbols = list(set(alert['symbol'] for alert in active_alerts))
# Lấy giá hiện tại từ Binance
current_prices = {}
for symbol in symbols:
params = {'symbol': symbol}
response = requests.get(BINANCE_API_URL, params=params)
data = response.json()
current_prices[symbol] = float(data['price'])
# Kiểm tra các điều kiện cảnh báo
triggered_alerts = []
for alert in active_alerts:
symbol = alert['symbol']
current_price = current_prices.get(symbol)
if not current_price:
continue
condition_met = False
# Kiểm tra điều kiện
if alert['condition'] == 'price_above' and current_price > alert['threshold']:
condition_met = True
elif alert['condition'] == 'price_below' and current_price < alert['threshold']:
condition_met = True
if condition_met:
triggered_alerts.append({
'alert_id': alert['alert_id'],
'user_id': alert['user_id'],
'symbol': symbol,
'condition': alert['condition'],
'threshold': alert['threshold'],
'current_price': current_price,
'contact_info': alert.get('contact_info', {})
})
# Gửi thông báo cho các cảnh báo đã kích hoạt
for alert in triggered_alerts:
# Tạo nội dung thông báo
message = f"CẢNH BÁO GIAO DỊCH: {alert['symbol']}\n"
message += f"Điều kiện: {alert['condition']} {alert['threshold']}\n"
message += f"Giá hiện tại: {alert['current_price']}"
# Gửi thông báo qua SNS
sns.publish(
TopicArn=SNS_TOPIC_ARN,
Message=message,
Subject=f"Cảnh báo giao dịch: {alert['symbol']}"
)
# Lưu lịch sử cảnh báo vào DynamoDB
history_table = dynamodb.Table('TradingAlertHistory')
history_table.put_item(
Item={
'alert_id': alert['alert_id'],
'timestamp': int(datetime.now().timestamp()),
'symbol': alert['symbol'],
'condition': alert['condition'],
'threshold': alert['threshold'],
'current_price': Decimal(str(alert['current_price']))
}
)
return {
'statusCode': 200,
'body': json.dumps({
'message': f"Kiểm tra hoàn tất. Đã kích hoạt {len(triggered_alerts)} cảnh báo."
})
}
except Exception as e:
print(f"Lỗi: {str(e)}")
return {
'statusCode': 500,
'body': json.dumps({
'message': f"Đã xảy ra lỗi: {str(e)}"
})
}
Bước 5: Tạo API Gateway để Kích Hoạt Lambda
import boto3
# Khởi tạo API Gateway client
apigateway = boto3.client('apigateway')
# Tạo REST API
api = apigateway.create_rest_api(
name='TradingAlertAPI',
description='API for trading alerts',
endpointConfiguration={
'types': ['REGIONAL']
}
)
# Lấy ID của API
api_id = api['id']
# Lấy ID của root resource
resources = apigateway.get_resources(restApiId=api_id)
root_id = [resource for resource in resources['items'] if resource['path'] == '/'][0]['id']
# Tạo resource /alerts
alerts_resource = apigateway.create_resource(
restApiId=api_id,
parentId=root_id,
pathPart='alerts'
)
alerts_id = alerts_resource['id']
# Tạo phương thức GET cho /alerts
apigateway.put_method(
restApiId=api_id,
resourceId=alerts_id,
httpMethod='GET',
authorizationType='NONE'
)
# Tạo phương thức POST cho /alerts
apigateway.put_method(
restApiId=api_id,
resourceId=alerts_id,
httpMethod='POST',
authorizationType='NONE'
)
# Giả sử chúng ta đã tạo Lambda function với tên TradingAlertProcessor
lambda_client = boto3.client('lambda')
lambda_function = lambda_client.get_function(FunctionName='TradingAlertProcessor')
lambda_arn = lambda_function['Configuration']['FunctionArn']
# Tích hợp Lambda với API Gateway
apigateway.put_integration(
restApiId=api_id,
resourceId=alerts_id,
httpMethod='GET',
type='AWS',
integrationHttpMethod='POST',
uri=f'arn:aws:apigateway:us-east-1:lambda:path/2015-03-31/functions/{lambda_arn}/invocations'
)
apigateway.put_integration(
restApiId=api_id,
resourceId=alerts_id,
httpMethod='POST',
type='AWS',
integrationHttpMethod='POST',
uri=f'arn:aws:apigateway:us-east-1:lambda:path/2015-03-31/functions/{lambda_arn}/invocations'
)
# Triển khai API
apigateway.create_deployment(
restApiId=api_id,
stageName='prod'
)
# In ra URL của API
api_url = f"https://{api_id}.execute-api.us-east-1.amazonaws.com/prod"
print(f"API URL: {api_url}")
Bước 6: Thiết lập Lịch Trình Chạy với CloudWatch Events
Để hệ thống kiểm tra giá định kỳ, chúng ta sẽ thiết lập một CloudWatch Event Rule:
import boto3
# Khởi tạo CloudWatch Events client
events = boto3.client('events')
# Tạo rule để chạy mỗi 5 phút
rule = events.put_rule(
Name='TradingAlertCheckSchedule',
ScheduleExpression='rate(5 minutes)',
State='ENABLED'
)
# Giả sử chúng ta đã tạo Lambda function với tên TradingAlertProcessor
lambda_client = boto3.client('lambda')
lambda_function = lambda_client.get_function(FunctionName='TradingAlertProcessor')
lambda_arn = lambda_function['Configuration']['FunctionArn']
# Thiết lập target cho rule
events.put_targets(
Rule='TradingAlertCheckSchedule',
Targets=[
{
'Id': '1',
'Arn': lambda_arn
}
]
)
# Cấp quyền cho CloudWatch Events gọi Lambda
lambda_client.add_permission(
FunctionName='TradingAlertProcessor',
StatementId='AllowCloudWatchEvents',
Action='lambda:InvokeFunction',
Principal='events.amazonaws.com',
SourceArn=rule['RuleArn']
)
print("Đã thiết lập lịch trình chạy mỗi 5 phút")
Bước 7: Xây Dựng Frontend Quản Lý Cảnh Báo
Bạn có thể xây dựng một frontend đơn giản sử dụng HTML, CSS, và JavaScript để quản lý cảnh báo. Dưới đây là một ví dụ HTML form cơ bản:
<!DOCTYPE html>
<html lang="vi">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Quản Lý Cảnh Báo Giao Dịch</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 0;
padding: 20px;
background-color: #f5f5f5;
}
.container {
max-width: 800px;
margin: 0 auto;
background-color: white;
padding: 20px;
border-radius: 5px;
box-shadow: 0 2px 5px rgba(0,0,0,0.1);
}
h1 {
color: #333;
}
.form-group {
margin-bottom: 15px;
}
label {
display: block;
margin-bottom: 5px;
font-weight: bold;
}
input, select {
width: 100%;
padding: 8px;
border: 1px solid #ddd;
border-radius: 4px;
box-sizing: border-box;
}
button {
background-color: #4CAF50;
color: white;
padding: 10px 15px;
border: none;
border-radius: 4px;
cursor: pointer;
}
button:hover {
background-color: #45a049;
}
.alert-list {
margin-top: 30px;
}
table {
width: 100%;
border-collapse: collapse;
}
th, td {
padding: 8px;
text-align: left;
border-bottom: 1px solid #ddd;
}
th {
background-color: #f2f2f2;
}
</style>
</head>
<body>
<div class="container">
<h1>Quản Lý Cảnh Báo Giao Dịch</h1>
<form id="alertForm">
<div class="form-group">
<label for="symbol">Mã Giao Dịch:</label>
<input type="text" id="symbol" name="symbol" placeholder="Ví dụ: BTCUSDT" required>
</div>
<div class="form-group">
<label for="condition">Điều Kiện:</label>
<select id="condition" name="condition" required>
<option value="price_above">Giá vượt trên</option>
<option value="price_below">Giá xuống dưới</option>
</select>
</div>
<div class="form-group">
<label for="threshold">Ngưỡng Giá:</label>
<input type="number" id="threshold" name="threshold" step="0.01" required>
</div>
<div class="form-group">
<label for="email">Email:</label>
<input type="email" id="email" name="email" required>
</div>
<div class="form-group">
<label for="phone">Số Điện Thoại (để nhận SMS):</label>
<input type="tel" id="phone" name="phone" placeholder="+84123456789">
</div>
<button type="submit">Tạo Cảnh Báo Mới</button>
</form>
<div class="alert-list">
<h2>Danh Sách Cảnh Báo</h2>
<table>
<thead>
<tr>
<th>Mã Giao Dịch</th>
<th>Điều Kiện</th>
<th>Ngưỡng Giá</th>
<th>Trạng Thái</th>
<th>Thao Tác</th>
</tr>
</thead>
<tbody id="alertsTableBody">
<!-- Dữ liệu sẽ được load động bằng JavaScript -->
</tbody>
</table>
</div>
</div>
<script>
// URL API của bạn
const API_URL = 'https://your-api-gateway-url.execute-api.us-east-1.amazonaws.com/prod/alerts';
// Lấy danh sách cảnh báo khi trang được load
document.addEventListener('DOMContentLoaded', fetchAlerts);
// Xử lý form submit
document.getElementById('alertForm').addEventListener('submit', function(e) {
e.preventDefault();
const alertData = {
symbol: document.getElementById('symbol').value,
condition: document.getElementById('condition').value,
threshold: parseFloat(document.getElementById('threshold').value),
contact_info: {
email: document.getElementById('email').value,
phone: document.getElementById('phone').value || null
},
active: true
};
createAlert(alertData);
});
// Hàm gọi API để tạo cảnh báo mới
async function createAlert(alertData) {
try {
const response = await fetch(API_URL, {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify(alertData)
});
if (!response.ok) {
throw new Error('Lỗi khi tạo cảnh báo');
}
const result = await response.json();
alert('Đã tạo cảnh báo thành công!');
// Refresh danh sách cảnh báo
fetchAlerts();
// Reset form
document.getElementById('alertForm').reset();
} catch (error) {
console.error('Lỗi:', error);
alert('Đã xảy ra lỗi: ' + error.message);
}
}
// Hàm gọi API để lấy danh sách cảnh báo
async function fetchAlerts() {
try {
const response = await fetch(API_URL);
if (!response.ok) {
throw new Error('Lỗi khi lấy danh sách cảnh báo');
}
const alerts = await response.json();
displayAlerts(alerts);
} catch (error) {
console.error('Lỗi:', error);
document.getElementById('alertsTableBody').innerHTML = `
<tr>
<td colspan="5">Không thể tải danh sách cảnh báo. Lỗi: ${error.message}</td>
</tr>
`;
}
}
// Hiển thị danh sách cảnh báo
function displayAlerts(alerts) {
const tableBody = document.getElementById('alertsTableBody');
if (!alerts || alerts.length === 0) {
tableBody.innerHTML = `
<tr>
<td colspan="5">Chưa có cảnh báo nào</td>
</tr>
`;
return;
}
let html = '';
alerts.forEach(alert => {
const condition = alert.condition === 'price_above' ? 'Giá vượt trên' : 'Giá xuống dưới';
const status = alert.active ? 'Hoạt động' : 'Tạm dừng';
html += `
<tr>
<td>${alert.symbol}</td>
<td>${condition}</td>
<td>${alert.threshold}</td>
<td>${status}</td>
<td>
<button onclick="toggleAlertStatus('${alert.alert_id}', ${!alert.active})">
${alert.active ? 'Tạm dừng' : 'Kích hoạt'}
</button>
<button onclick="deleteAlert('${alert.alert_id}')">Xóa</button>
</td>
</tr>
`;
});
tableBody.innerHTML = html;
}
// Hàm để bật/tắt trạng thái cảnh báo
async function toggleAlertStatus(alertId, newActiveStatus) {
try {
const response = await fetch(`${API_URL}/${alertId}`, {
method: 'PATCH',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({
active: newActiveStatus
})
});
if (!response.ok) {
throw new Error('Lỗi khi cập nhật trạng thái cảnh báo');
}
alert('Đã cập nhật trạng thái cảnh báo');
fetchAlerts();
} catch (error) {
console.error('Lỗi:', error);
alert('Đã xảy ra lỗi: ' + error.message);
}
}
// Hàm để xóa cảnh báo
async function deleteAlert(alertId) {
if (!confirm('Bạn có chắc chắn muốn xóa cảnh báo này?')) {
return;
}
try {
const response = await fetch(`${API_URL}/${alertId}`, {
method: 'DELETE'
});
if (!response.ok) {
throw new Error('Lỗi khi xóa cảnh báo');
}
alert('Đã xóa cảnh báo thành công');
fetchAlerts();
} catch (error) {
console.error('Lỗi:', error);
alert('Đã xảy ra lỗi: ' + error.message);
}
}
</script>
</body>
</html>
Bước 8: Triển Khai và Kiểm Tra
Triển khai hàm Lambda:
- Tải mã nguồn lên S3 bucket
- Tạo Lambda function và cấu hình
Kiểm tra hệ thống:
- Tạo một cảnh báo thử nghiệm
- Kích hoạt bằng tay hoặc đợi trigger theo lịch
- Kiểm tra thông báo qua các kênh đã cấu hình
Mở Rộng Hệ Thống
Sau khi đã triển khai thành công hệ thống cơ bản, bạn có thể mở rộng nó bằng các tính năng sau:
1. Thêm Các Loại Điều Kiện Phức Tạp Hơn
# Ví dụ về hàm kiểm tra điều kiện phức tạp hơn
def check_complex_conditions(alert, current_data, historical_data):
if alert['condition_type'] == 'moving_average_cross':
short_ma = calculate_moving_average(historical_data, alert['short_period'])
long_ma = calculate_moving_average(historical_data, alert['long_period'])
if alert['cross_direction'] == 'above' and short_ma > long_ma:
return True
elif alert['cross_direction'] == 'below' and short_ma < long_ma:
return True
elif alert['condition_type'] == 'rsi_level':
rsi = calculate_rsi(historical_data, alert['rsi_period'])
if alert['rsi_direction'] == 'overbought' and rsi > alert['rsi_threshold']:
return True
elif alert['rsi_direction'] == 'oversold' and rsi < alert['rsi_threshold']:
return True
return False
2. Phân Tích Dữ Liệu Lịch Sử Cảnh Báo
import pandas as pd
import matplotlib.pyplot as plt
import boto3
from io import BytesIO
import base64
def generate_alert_analytics(user_id):
# Lấy lịch sử cảnh báo từ DynamoDB
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('TradingAlertHistory')
response = table.query(
IndexName='UserIdIndex',
KeyConditionExpression='user_id = :uid',
ExpressionAttributeValues={':uid': user_id}
)
alert_history = response['Items']
# Chuyển đổi sang DataFrame
df = pd.DataFrame(alert_history)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')
# Phân tích theo symbol
symbol_counts = df['symbol'].value_counts()
# Tạo biểu đồ
plt.figure(figsize=(10, 6))
symbol_counts.plot(kind='bar')
plt.title('Số Lượng Cảnh Báo Theo Mã Giao Dịch')
plt.xlabel('Mã Giao Dịch')
plt.ylabel('Số Lượng Cảnh Báo')
plt.tight_layout()
# Lưu biểu đồ vào buffer
buffer = BytesIO()
plt.savefig(buffer, format='png')
buffer.seek(0)
# Chuyển đổi thành base64 để hiển thị trên web
image_base64 = base64.b64encode(buffer.read()).decode('utf-8')
return {
'total_alerts': len(alert_history),
'most_common_symbol': symbol_counts.index[0],
'chart_data': image_base64
}
3. Thêm Xác Thực và Ủy Quyền
import boto3
import json
import os
import time
import hashlib
import hmac
import base64
from jose import jwt
def create_cognito_user_pool():
cognito = boto3.client('cognito-idp')
response = cognito.create_user_pool(
PoolName='TradingAlertUsers',
Policies={
'PasswordPolicy': {
'MinimumLength': 8,
'RequireUppercase': True,
'RequireLowercase': True,
'RequireNumbers': True,
'RequireSymbols': False
}
},
AutoVerifiedAttributes=['email']
)
user_pool_id = response['UserPool']['Id']
# Tạo client app
client_response = cognito.create_user_pool_client(
UserPoolId=user_pool_id,
ClientName='trading-alert-client',
GenerateSecret=False,
ExplicitAuthFlows=['ALLOW_USER_PASSWORD_AUTH', 'ALLOW_REFRESH_TOKEN_AUTH']
)
client_id = client_response['UserPoolClient']['ClientId']
print(f"User Pool ID: {user_pool_id}")
print(f"Client ID: {client_id}")
return user_pool_id, client_id
Ước Tính Chi Phí
AWS Lambda có lớp miễn phí rộng rãi, nhưng để giúp bạn lập kế hoạch ngân sách, dưới đây là ước tính chi phí hàng tháng cho hệ thống cảnh báo với quy mô vừa phải:
AWS Lambda:
- Miễn phí: 1 triệu yêu cầu/tháng và 400.000 GB-giây/tháng
- Nếu vượt quá: $0.20 cho mỗi 1 triệu yêu cầu
API Gateway:
- $3.50 cho mỗi triệu API calls
DynamoDB:
- Miễn phí: 25GB lưu trữ, 25 đơn vị đọc/ghi
- Chi phí bổ sung thay đổi theo mức sử dụng
SNS:
- $0.50 cho mỗi triệu thông báo
- $0.75 cho 100 SMS
CloudWatch:
- $0.30 cho mỗi GB dữ liệu log
- $0.10 cho mỗi alarm/tháng
Tổng ước tính cho quy mô nhỏ đến vừa: $10-$30/tháng
Kết luận
Trong bài viết này, chúng ta đã xây dựng một hệ thống cảnh báo giao dịch toàn diện sử dụng AWS Lambda và các dịch vụ AWS khác. Hệ thống này có thể dễ dàng mở rộng để hỗ trợ hàng ngàn người dùng và các điều kiện giao dịch phức tạp.
Các lợi ích chính của giải pháp này:
- Khả năng mở rộng tự động: Tự động mở rộng theo khối lượng yêu cầu
- Chi phí thấp: Chỉ trả tiền cho những gì bạn sử dụng
- Độ tin cậy cao: Sử dụng các dịch vụ được quản lý của AWS
- Dễ bảo trì: Kiến trúc serverless giảm thiểu nhu cầu quản lý hạ tầng
Hãy tùy chỉnh hệ thống theo nhu cầu cụ thể của bạn và mở rộng nó với các tính năng phân tích dữ liệu, điều kiện giao dịch phức tạp hơn, hoặc tích hợp với các nền tảng giao dịch tự động.