yeshwanth23/mlops
0
1#!/usr/bin/env python3
2"""
3Plotly Dash Dashboard for Bitcoin Transaction Anomaly Detection
4This script creates an interactive dashboard for visualizing Bitcoin transaction data
5and anomaly detection results.
6"""
7
8import os
9import json
10import pandas as pd
11import numpy as np
12import requests
13from datetime import datetime, timedelta
14
15import dash
16from dash import dcc, html, Input, Output, State, callback
17import dash_bootstrap_components as dbc
18import plotly.express as px
19import plotly.graph_objects as go
20
21# API endpoint
22API_ENDPOINT = "http://localhost:5000/predict"
23
24# Create Dash app
25app = dash.Dash(
26 __name__,
27 external_stylesheets=[dbc.themes.BOOTSTRAP],
28 meta_tags=[{"name": "viewport", "content": "width=device-width, initial-scale=1"}],
29)
30
31app.title = "Bitcoin Transaction Anomaly Detection"
32
33# Load sample data (in a real scenario, this would come from Delta Lake)
34def load_sample_data():
35 """
36 Load sample Bitcoin transaction data
37 In a real implementation, this would load data from Delta Lake
38 """
39 # Generate synthetic data for demonstration
40 np.random.seed(42)
41 n_samples = 1000
42
43 # Generate timestamps for the last 24 hours
44 now = datetime.now()
45 timestamps = [now - timedelta(minutes=np.random.randint(0, 24*60)) for _ in range(n_samples)]
46 timestamps.sort()
47
48 # Generate transaction data
49 data = {
50 'hash': [f"tx_{i}" for i in range(n_samples)],
51 'transaction_time': timestamps,
52 'size': np.random.lognormal(7, 1, n_samples).astype(int),
53 'weight': np.random.lognormal(8, 1, n_samples).astype(int),
54 'fee': np.random.lognormal(9, 1.5, n_samples).astype(int),
55 'inputs_count': np.random.randint(1, 10, n_samples),
56 'outputs_count': np.random.randint(1, 5, n_samples),
57 'input_value': np.random.lognormal(16, 2, n_samples).astype(int),
58 'output_value': np.random.lognormal(16, 2, n_samples).astype(int),
59 }
60
61 # Calculate derived metrics
62 df = pd.DataFrame(data)
63 df['fee_rate'] = df['fee'] / df['size']
64 df['fee_per_weight'] = df['fee'] / df['weight']
65
66 # Generate anomaly scores (mostly normal with a few anomalies)
67 anomaly_scores = np.random.normal(0, 0.5, n_samples)
68 # Make a few transactions anomalous
69 anomaly_indices = np.random.choice(n_samples, size=int(n_samples * 0.02), replace=False)
70 for idx in anomaly_indices:
71 anomaly_scores[idx] = np.random.uniform(1.5, 3)
72
73 df['anomaly_score'] = anomaly_scores
74 df['is_anomaly'] = df['anomaly_score'] > 1.0
75
76 return df
77
78# App layout
79app.layout = dbc.Container(
80 [
81 # Header
82 dbc.Row(
83 [
84 dbc.Col(
85 [
86 html.H1("Bitcoin Transaction Anomaly Detection", className="display-4"),
87 html.P(
88 "Interactive dashboard for monitoring Bitcoin transactions and detecting anomalies",
89 className="lead",
90 ),
91 ],
92 width={"size": 10, "offset": 1},
93 )
94 ],
95 className="mb-4 mt-4",
96 ),
97
98 # Filters and Controls
99 dbc.Row(
100 [
101 dbc.Col(
102 [
103 dbc.Card(
104 [
105 dbc.CardHeader("Filters"),
106 dbc.CardBody(
107 [
108 dbc.Row(
109 [
110 dbc.Col(
111 [
112 html.Label("Time Range"),
113 dcc.Dropdown(
114 id="time-range-dropdown",
115 options=[
116 {"label": "Last Hour", "value": "1h"},
117 {"label": "Last 6 Hours", "value": "6h"},
118 {"label": "Last 12 Hours", "value": "12h"},
119 {"label": "Last 24 Hours", "value": "24h"},
120 {"label": "All Time", "value": "all"},
121 ],
122 value="24h",
123 ),
124 ],
125 width=6,
126 ),
127 dbc.Col(
128 [
129 html.Label("Anomaly Threshold"),
130 dcc.Slider(
131 id="anomaly-threshold-slider",
132 min=0,
133 max=2,
134 step=0.1,
135 value=1.0,
136 marks={i: str(i) for i in range(0, 3)},
137 ),
138 ],
139 width=6,
140 ),
141 ]
142 ),
143 html.Br(),
144 dbc.Row(
145 [
146 dbc.Col(
147 [
148 html.Label("Transaction Size Range (bytes)"),
149 dcc.RangeSlider(
150 id="size-range-slider",
151 min=0,
152 max=10000,
153 step=100,
154 value=[0, 10000],
155 marks={i: str(i) for i in range(0, 10001, 2000)},
156 ),
157 ]
158 )
159 ]
160 ),
161 html.Br(),
162 dbc.Row(
163 [
164 dbc.Col(
165 [
166 html.Label("Show Anomalies Only"),
167 dbc.Switch(id="anomalies-only-switch", value=False),
168 ],
169 width=6,
170 ),
171 dbc.Col(
172 [
173 html.Label("Auto Refresh"),
174 dbc.Switch(id="auto-refresh-switch", value=True),
175 ],
176 width=6,
177 ),
178 ]
179 ),
180 ]
181 ),
182 ],
183 className="mb-4",
184 ),
185 ],
186 width={"size": 10, "offset": 1},
187 )
188 ]
189 ),
190
191 # KPI Cards
192 dbc.Row(
193 [
194 dbc.Col(
195 [
196 dbc.Card(
197 [
198 dbc.CardBody(
199 [
200 html.H4("Total Transactions", className="card-title"),
201 html.H2(id="total-transactions", className="card-value"),
202 ]
203 )
204 ],
205 className="mb-4 text-center",
206 ),
207 ],
208 width=3,
209 ),
210 dbc.Col(
211 [
212 dbc.Card(
213 [
214 dbc.CardBody(
215 [
216 html.H4("Anomalies Detected", className="card-title"),
217 html.H2(id="total-anomalies", className="card-value text-danger"),
218 ]
219 )
220 ],
221 className="mb-4 text-center",
222 ),
223 ],
224 width=3,
225 ),
226 dbc.Col(
227 [
228 dbc.Card(
229 [
230 dbc.CardBody(
231 [
232 html.H4("Avg. Fee Rate (sat/byte)", className="card-title"),
233 html.H2(id="avg-fee-rate", className="card-value"),
234 ]
235 )
236 ],
237 className="mb-4 text-center",
238 ),
239 ],
240 width=3,
241 ),
242 dbc.Col(
243 [
244 dbc.Card(
245 [
246 dbc.CardBody(
247 [
248 html.H4("Avg. Transaction Size", className="card-title"),
249 html.H2(id="avg-tx-size", className="card-value"),
250 ]
251 )
252 ],
253 className="mb-4 text-center",
254 ),
255 ],
256 width=3,
257 ),
258 ],
259 className="mb-4",
260 ),
261
262 # Charts
263 dbc.Row(
264 [
265 dbc.Col(
266 [
267 dbc.Card(
268 [
269 dbc.CardHeader("Transactions Over Time"),
270 dbc.CardBody(
271 [
272 dcc.Graph(id="transactions-time-chart"),
273 ]
274 ),
275 ],
276 className="mb-4",
277 ),
278 ],
279 width=6,
280 ),
281 dbc.Col(
282 [
283 dbc.Card(
284 [
285 dbc.CardHeader("Anomaly Score Distribution"),
286 dbc.CardBody(
287 [
288 dcc.Graph(id="anomaly-score-histogram"),
289 ]
290 ),
291 ],
292 className="mb-4",
293 ),
294 ],
295 width=6,
296 ),
297 ]
298 ),
299
300 dbc.Row(
301 [
302 dbc.Col(
303 [
304 dbc.Card(
305 [
306 dbc.CardHeader("Fee Rate vs. Transaction Size"),
307 dbc.CardBody(
308 [
309 dcc.Graph(id="fee-size-scatter"),
310 ]
311 ),
312 ],
313 className="mb-4",
314 ),
315 ],
316 width=12,
317 ),
318 ]
319 ),
320
321 # Transaction Table
322 dbc.Row(
323 [
324 dbc.Col(
325 [
326 dbc.Card(
327 [
328 dbc.CardHeader("Recent Transactions"),
329 dbc.CardBody(
330 [
331 html.Div(id="transactions-table"),
332 ]
333 ),
334 ],
335 className="mb-4",
336 ),
337 ],
338 width=12,
339 ),
340 ]
341 ),
342
343 # Anomaly Prediction Form
344 dbc.Row(
345 [
346 dbc.Col(
347 [
348 dbc.Card(
349 [
350 dbc.CardHeader("Test Transaction Anomaly Detection"),
351 dbc.CardBody(
352 [
353 dbc.Row(
354 [
355 dbc.Col(
356 [
357 html.Label("Transaction Hash"),
358 dbc.Input(id="tx-hash-input", placeholder="Enter transaction hash", type="text"),
359 ],
360 width=6,
361 ),
362 dbc.Col(
363 [
364 html.Label("Transaction Size (bytes)"),
365 dbc.Input(id="tx-size-input", placeholder="Enter size", type="number", value=250),
366 ],
367 width=3,
368 ),
369 dbc.Col(
370 [
371 html.Label("Transaction Weight"),
372 dbc.Input(id="tx-weight-input", placeholder="Enter weight", type="number", value=1000),
373 ],
374 width=3,
375 ),
376 ]
377 ),
378 html.Br(),
379 dbc.Row(
380 [
381 dbc.Col(
382 [
383 html.Label("Fee (satoshis)"),
384 dbc.Input(id="tx-fee-input", placeholder="Enter fee", type="number", value=5000),
385 ],
386 width=3,
387 ),
388 dbc.Col(
389 [
390 html.Label("Inputs Count"),
391 dbc.Input(id="tx-inputs-input", placeholder="Enter inputs count", type="number", value=2),
392 ],
393 width=3,
394 ),
395 dbc.Col(
396 [
397 html.Label("Outputs Count"),
398 dbc.Input(id="tx-outputs-input", placeholder="Enter outputs count", type="number", value=2),
399 ],
400 width=3,
401 ),
402 dbc.Col(
403 [
404 html.Label("Input Value (satoshis)"),
405 dbc.Input(id="tx-input-value-input", placeholder="Enter input value", type="number", value=1000000),
406 ],
407 width=3,
408 ),
409 ]
410 ),
411 html.Br(),
412 dbc.Row(
413 [
414 dbc.Col(
415 [
416 html.Label("Output Value (satoshis)"),
417 dbc.Input(id="tx-output-value-input", placeholder="Enter output value", type="number", value=995000),
418 ],
419 width=3,
420 ),
421 dbc.Col(
422 [
423 dbc.Button("Check Transaction", id="check-tx-button", color="primary", className="mt-4"),
424 ],
425 width=3,
426 ),
427 dbc.Col(
428 [
429 html.Div(id="prediction-result", className="mt-4"),
430 ],
431 width=6,
432 ),
433 ]
434 ),
435 ]
436 ),
437 ],
438 className="mb-4",
439 ),
440 ],
441 width={"size": 10, "offset": 1},
442 )
443 ]
444 ),
445
446 # Footer
447 dbc.Row(
448 [
449 dbc.Col(
450 [
451 html.Hr(),
452 html.P(
453 "Bitcoin Transaction Anomaly Detection Dashboard - Powered by MLOps Pipeline",
454 className="text-center text-muted",
455 ),
456 ],
457 width=12,
458 )
459 ]
460 ),
461
462 # Store for data
463 dcc.Store(id="transaction-data-store"),
464
465 # Interval for auto refresh
466 dcc.Interval(
467 id="auto-refresh-interval",
468 interval=30 * 1000, # 30 seconds
469 n_intervals=0,
470 disabled=False,
471 ),
472 ],
473 fluid=True,
474)
475
476# Callbacks
477@app.callback(
478 Output("transaction-data-store", "data"),
479 [
480 Input("auto-refresh-interval", "n_intervals"),
481 Input("anomaly-threshold-slider", "value"),
482 Input("size-range-slider", "value"),
483 Input("time-range-dropdown", "value"),
484 ],
485)
486def update_data(n_intervals, anomaly_threshold, size_range, time_range):
487 """
488 Update the transaction data based on filters
489 """
490 # Load sample data
491 df = load_sample_data()
492
493 # Apply time range filter
494 now = datetime.now()
495 if time_range != "all":
496 hours = int(time_range.replace("h", ""))
497 df = df[df["transaction_time"] >= now - timedelta(hours=hours)]
498
499 # Apply size range filter
500 df = df[(df["size"] >= size_range[0]) & (df["size"] <= size_range[1])]
501
502 # Update anomaly flag based on threshold
503 df["is_anomaly"] = df["anomaly_score"] > anomaly_threshold
504
505 # Convert to JSON for storage
506 return df.to_json(date_format="iso", orient="split")
507
508@app.callback(
509 [
510 Output("total-transactions", "children"),
511 Output("total-anomalies", "children"),
512 Output("avg-fee-rate", "children"),
513 Output("avg-tx-size", "children"),
514 ],
515 [
516 Input("transaction-data-store", "data"),
517 Input("anomalies-only-switch", "value"),
518 ],
519)
520def update_kpi_cards(json_data, anomalies_only):
521 """
522 Update KPI cards based on filtered data
523 """
524 df = pd.read_json(json_data, orient="split")
525
526 if anomalies_only:
527 df = df[df["is_anomaly"]]
528
529 total_tx = len(df)
530 total_anomalies = df["is_anomaly"].sum()
531 avg_fee_rate = df["fee_rate"].mean()
532 avg_tx_size = df["size"].mean()
533
534 return (
535 f"{total_tx:,}",
536 f"{total_anomalies:,}",
537 f"{avg_fee_rate:.2f}",
538 f"{avg_tx_size:,.0f}",
539 )
540
541@app.callback(
542 Output("transactions-time-chart", "figure"),
543 [
544 Input("transaction-data-store", "data"),
545 Input("anomalies-only-switch", "value"),
546 Input("anomaly-threshold-slider", "value"),
547 ],
548)
549def update_time_chart(json_data, anomalies_only, anomaly_threshold):
550 """
551 Update transactions over time chart
552 """
553 df = pd.read_json(json_data, orient="split")
554
555 if anomalies_only:
556 df = df[df["is_anomaly"]]
557
558 # Group by hour
559 df["hour"] = df["transaction_time"].dt.floor("H")
560 hourly_counts = df.groupby(["hour", "is_anomaly"]).size().reset_index(name="count")
561
562 # Create figure
563 fig = go.Figure()
564
565 # Add normal transactions
566 normal_data = hourly_counts[~hourly_counts["is_anomaly"]]
567 if not normal_data.empty:
568 fig.add_trace(
569 go.Scatter(
570 x=normal_data["hour"],
571 y=normal_data["count"],
572 mode="lines",
573 name="Normal Transactions",
574 line=dict(color="green", width=2),
575 stackgroup="one",
576 )
577 )
578
579 # Add anomalous transactions
580 anomaly_data = hourly_counts[hourly_counts["is_anomaly"]]
581 if not anomaly_data.empty:
582 fig.add_trace(
583 go.Scatter(
584 x=anomaly_data["hour"],
585 y=anomaly_data["count"],
586 mode="lines",
587 name="Anomalous Transactions",
588 line=dict(color="red", width=2),
589 stackgroup="one",
590 )
591 )
592
593 fig.update_layout(
594 title="Transaction Volume Over Time",
595 xaxis_title="Time",
596 yaxis_title="Number of Transactions",
597 legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
598 margin=dict(l=40, r=40, t=40, b=40),
599 )
600
601 return fig
602
603@app.callback(
604 Output("anomaly-score-histogram", "figure"),
605 [
606 Input("transaction-data-store", "data"),
607 Input("anomaly-threshold-slider", "value"),
608 ],
609)
610def update_anomaly_histogram(json_data, anomaly_threshold):
611 """
612 Update anomaly score histogram
613 """
614 df = pd.read_json(json_data, orient="split")
615
616 fig = go.Figure()
617
618 # Add histogram
619 fig.add_trace(
620 go.Histogram(
621 x=df["anomaly_score"],
622 nbinsx=30,
623 marker_color="lightblue",
624 )
625 )
626
627 # Add threshold line
628 fig.add_shape(
629 type="line",
630 x0=anomaly_threshold,
631 x1=anomaly_threshold,
632 y0=0,
633 y1=1,
634 yref="paper",
635 line=dict(color="red", width=2, dash="dash"),
636 )
637
638 fig.add_annotation(
639 x=anomaly_threshold,
640 y=0.95,
641 yref="paper",
642 text="Threshold",
643 showarrow=True,
644 arrowhead=1,
645 ax=40,
646 ay=0,
647 )
648
649 fig.update_layout(
650 title="Distribution of Anomaly Scores",
651 xaxis_title="Anomaly Score",
652 yaxis_title="Count",
653 margin=dict(l=40, r=40, t=40, b=40),
654 )
655
656 return fig
657
658@app.callback(
659 Output("fee-size-scatter", "figure"),
660 [
661 Input("transaction-data-store", "data"),
662 Input("anomalies-only-switch", "value"),
663 ],
664)
665def update_fee_size_scatter(json_data, anomalies_only):
666 """
667 Update fee rate vs. transaction size scatter plot
668 """
669 df = pd.read_json(json_data, orient="split")
670
671 if anomalies_only:
672 df = df[df["is_anomaly"]]
673
674 fig = px.scatter(
675 df,
676 x="size",
677 y="fee_rate",
678 color="is_anomaly",
679 color_discrete_map={True: "red", False: "blue"},
680 hover_data=["hash", "fee", "inputs_count", "outputs_count"],
681 opacity=0.7,
682 title="Fee Rate vs. Transaction Size",
683 )
684
685 fig.update_layout(
686 xaxis_title="Transaction Size (bytes)",
687 yaxis_title="Fee Rate (satoshis/byte)",
688 legend_title="Is Anomaly",
689 margin=dict(l=40, r=40, t=40, b=40),
690 )
691
692 return fig
693
694@app.callback(
695 Output("transactions-table", "children"),
696 [
697 Input("transaction-data-store", "data"),
698 Input("anomalies-only-switch", "value"),
699 ],
700)
701def update_transactions_table(json_data, anomalies_only):
702 """
703 Update transactions table
704 """
705 df = pd.read_json(json_data, orient="split")
706
707 if anomalies_only:
708 df = df[df["is_anomaly"]]
709
710 # Sort by time (most recent first) and take the most recent 10
711 df = df.sort_values("transaction_time", ascending=False).head(10)
712
713 # Format the table
714 table_header = [
715 html.Thead(
716 html.Tr(
717 [
718 html.Th("Time"),
719 html.Th("Hash"),
720 html.Th("Size"),
721 html.Th("Fee"),
722 html.Th("Fee Rate"),
723 html.Th("Anomaly Score"),
724 html.Th("Status"),
725 ]
726 )
727 )
728 ]
729
730 rows = []
731 for _, row in df.iterrows():
732 status_badge = dbc.Badge("Anomaly", color="danger") if row["is_anomaly"] else dbc.Badge("Normal", color="success")
733
734 rows.append(
735 html.Tr(
736 [
737 html.Td(row["transaction_time"].strftime("%Y-%m-%d %H:%M:%S")),
738 html.Td(row["hash"][:10] + "..."),
739 html.Td(f"{row['size']:,}"),
740 html.Td(f"{row['fee']:,}"),
741 html.Td(f"{row['fee_rate']:.2f}"),
742 html.Td(f"{row['anomaly_score']:.2f}"),
743 html.Td(status_badge),
744 ]
745 )
746 )
747
748 table_body = [html.Tbody(rows)]
749
750 return dbc.Table(table_header + table_body, bordered=True, hover=True, responsive=True, striped=True)
751
752@app.callback(
753 Output("auto-refresh-interval", "disabled"),
754 [Input("auto-refresh-switch", "value")],
755)
756def toggle_auto_refresh(auto_refresh):
757 """
758 Toggle auto refresh
759 """
760 return not auto_refresh
761
762@app.callback(
763 Output("prediction-result", "children"),
764 [Input("check-tx-button", "n_clicks")],
765 [
766 State("tx-hash-input", "value"),
767 State("tx-size-input", "value"),
768 State("tx-weight-input", "value"),
769 State("tx-fee-input", "value"),
770 State("tx-inputs-input", "value"),
771 State("tx-outputs-input", "value"),
772 State("tx-input-value-input", "value"),
773 State("tx-output-value-input", "value"),
774 ],
775)
776def check_transaction(n_clicks, tx_hash, size, weight, fee, inputs, outputs, input_value, output_value):
777 """
778 Check a transaction for anomalies
779 """
780 if n_clicks is None:
781 return ""
782
783 # In a real implementation, this would call the API
784 # For demonstration, we'll simulate a response
785
786 # Create transaction data
787 tx_data = {
788 "hash": tx_hash or "test_transaction",
789 "size": size or 250,
790 "weight": weight or 1000,
791 "fee": fee or 5000,
792 "inputs_count": inputs or 2,
793 "outputs_count": outputs or 2,
794 "input_value": input_value or 1000000,
795 "output_value": output_value or 995000,
796 }
797
798 try:
799 # In a real implementation, this would be an API call
800 # response = requests.post(API_ENDPOINT, json=tx_data)
801 # result = response.json()
802
803 # For demonstration, simulate a response
804 fee_rate = tx_data["fee"] / tx_data["size"]
805 is_anomaly = False
806 explanation = "Transaction appears normal based on its characteristics."
807
808 # Simple anomaly detection logic for demonstration
809 if fee_rate > 100 or fee_rate < 1 or tx_data["size"] > 10000:
810 is_anomaly = True
811 explanation = f"Potential anomaly detected: Unusual fee rate ({fee_rate:.2f} satoshis/byte)"
812
813 result = {
814 "transaction_hash": tx_data["hash"],
815 "anomaly_score": 1.5 if is_anomaly else 0.2,
816 "is_anomaly": is_anomaly,
817 "explanation": explanation,
818 }
819
820 # Create result display
821 status_badge = dbc.Badge("ANOMALY DETECTED", color="danger") if result["is_anomaly"] else dbc.Badge("NORMAL", color="success")
822
823 return html.Div(
824 [
825 html.Div([status_badge], className="mb-2"),
826 html.Div(f"Anomaly Score: {result['anomaly_score']:.2f}", className="mb-2"),
827 html.Div(result["explanation"]),
828 ]
829 )
830
831 except Exception as e:
832 return html.Div(f"Error: {str(e)}", className="text-danger")
833
834if __name__ == "__main__":
835 app.run_server(debug=True, host="0.0.0.0", port=8050)
836