import gradio as gr
import pandas as pd
import plotly.graph_objects as go
from datetime import datetime
import os
from preprocessing import get_latest_sequence
from predictor import MineROIPredictor
from fetch_blockchain_data import get_latest_blockchain_data
# internal dummy miner used only for age_days etc. (not shown in UI)
DEFAULT_MINER_NAME = "s19pro"
MODEL_PATH = "best_model_weights.pth"
# Predictor (global)
predictor = None
def init_predictor():
"""Initialize predictor once"""
global predictor
if predictor is None:
predictor = MineROIPredictor(MODEL_PATH)
def init_app():
"""Initialize app (no need for local blockchain_data_complete.csv)."""
print("\n" + "="*80)
print("š INITIALIZING MINEROI-NET APP")
print("="*80)
print("\nUsing live blockchain.com data (last 90 days).")
print("Model will use the latest 30 days for ROI prediction.")
print("="*80 + "\n")
init_predictor()
def predict_roi(machine_price, machine_hashrate, machine_power, machine_efficiency, electricity_rate,machine_release_date):
"""
Real-time prediction:
- Uses latest 90 days from blockchain.com
- Model uses last 30 days
- Scaler chosen based on electricity_rate:
< 0.05 -> ethiopia scaler
0.05-0.09 -> china scaler
> 0.09 -> texas scaler
"""
try:
window_size = 30
# -------- parse user inputs --------
miner_price = float(machine_price)
miner_hashrate = float(machine_hashrate)
machine_power = float(machine_power)
machine_efficiency = float(machine_efficiency)
user_electricity_rate = float(electricity_rate)
# ----- parse release date -----
release_str = None
if machine_release_date is not None:
release_str = str(machine_release_date).strip()
if release_str:
try:
# validate format YYYY-MM-DD
datetime.strptime(release_str, "%Y-%m-%d")
except ValueError:
error_msg = """
ā Invalid release date
Please enter the machine release date in the format YYYY-MM-DD,
for example 2020-05-01.
"""
return error_msg, error_msg, None, None
else:
# empty box -> fall back to S19 Pro default
release_str = None
# ---------------------------------------------------------
# Bucket electricity rate ONLY to choose scaler
# The actual feature value will be the user input, repeated
# for all 30 days in the window.
# ---------------------------------------------------------
if user_electricity_rate < 0.05:
scaler_region = "ethiopia"
region_bucket = "Low-cost (< $0.05/kWh)"
elif user_electricity_rate < 1.0:
scaler_region = "china"
region_bucket = "Medium-cost ($0.05ā$1.00/kWh)"
else:
scaler_region = "texas"
region_bucket = "High-cost (ā„ $1.00/kWh)"
# Region for the pipeline (used by get_latest_sequence & predictor)
region = scaler_region
# This is the value that will be used for all 30 days in the window
# (prepare_miner_features repeats it across time).
electricity_rate_used = user_electricity_rate
print("User machine specs:")
print(f" Price: {miner_price}")
print(f" Hashrate (TH/s): {miner_hashrate}")
print(f" Power (W): {machine_power}")
print(f" Efficiency: {machine_efficiency}")
print(f" User elec rate: {user_electricity_rate} USD/kWh")
print(f" Bucket region: {region}")
print(f" Elec used in features (all 30 days): {electricity_rate_used} USD/kWh")
print("=" * 80 + "\n")
# # -------- choose scaler region from electricity_rate --------
# if electricity_rate < 0.05:
# scaler_region = "texas"
# region_bucket = "Low-cost (< $0.05/kWh)"
# elif electricity_rate <= 0.09:
# scaler_region = "texas"
# region_bucket = "Medium-cost ($0.05ā$0.09/kWh)"
# else:
# scaler_region = "texas"
# region_bucket = "High-cost (> $0.09/kWh)"
# print("\n" + "=" * 80)
# print("PREDICTION REQUEST")
# print("=" * 80)
# print(f"Scaler region (from electricity rate): {scaler_region}")
print("=" * 80 + "\n")
# -------- fetch latest blockchain data (no date input) --------
print("š” Fetching latest blockchain data (last 90 days)...")
blockchain_df = get_latest_blockchain_data(days=90)
if blockchain_df is None or len(blockchain_df) < window_size:
error_msg = f"""
ā Error: Insufficient Data
Not enough blockchain data available.
Need at least {window_size} days of historical data.
"""
return error_msg, error_msg, None, None
print(f"ā
Got {len(blockchain_df)} days of data")
print(f" Date range: {blockchain_df['date'].min().date()} to {blockchain_df['date'].max().date()}")
price_source = "User input"
print(f" Using user-provided price: ${miner_price:,.2f}")
# -------- build sequence with user machine specs --------
print("\nš§ Preparing features...")
sequence, df_window, pred_date = get_latest_sequence(
blockchain_df,
DEFAULT_MINER_NAME, # internal dummy miner, not shown to user
miner_price,
scaler_region,
window_size,
machine_hashrate=miner_hashrate,
power=machine_power,
efficiency=machine_efficiency,
electricity_rate=electricity_rate_used,
release_date=release_str
)
print(f"ā
Sequence prepared: {sequence.shape}")
# -------- model prediction --------
print("\nš¤ Running prediction...")
result = predictor.predict(sequence, scaler_region)
print(f"ā
Prediction: {result['predicted_label']} ({result['confidence']:.1%})")
# -------- build UI outputs --------
miner_info = create_miner_info(
miner_price,
price_source,
pred_date,
miner_hashrate,
machine_power,
machine_efficiency,
electricity_rate_used,
region_bucket,
)
prediction_html = create_prediction_html(result, pred_date, window_size)
confidence_chart = create_confidence_chart(result["probabilities"])
price_chart = create_price_chart(blockchain_df, window_size)
print("=" * 80 + "\n")
return miner_info, prediction_html, confidence_chart, price_chart
except Exception as e:
import traceback
error_details = traceback.format_exc()
print("\nā ERROR:")
print(error_details)
error = f"""
ā Prediction Error
{str(e)}
"""
return error, error, None, None
def create_miner_info(
price,
source,
pred_date,
machine_hashrate,
machine_power,
machine_efficiency,
electricity_rate,
region_bucket,
):
"""
Display miner info for a user-specified machine (no ASIC dropdown).
"""
elec_rate = float(electricity_rate)
daily_cost = (float(machine_power) * 24.0 / 1000.0) * elec_rate
# Color coding for price source
if source == "API":
badge_color = "#27ae60" # Green
elif source == "User input":
badge_color = "#3498db" # Blue
else:
badge_color = "#e74c3c" # Red
return f"""
Custom ASIC Miner
Hashrate:{machine_hashrate:.2f} TH/s
Power:{machine_power:.1f} W
Efficiency:{machine_efficiency:.2f} W/TH
Price (as of {pred_date.date()}):${price:,.2f}
{source}
Electricity rate:{elec_rate:.4f} USD/kWh
Cost bucket:{region_bucket}
Estimated daily elec cost:${daily_cost:,.2f}
"""
def create_prediction_html(result, date, window):
label = result['predicted_label']
conf = result['confidence']
if 'Unprofitable' in label:
color, emoji, rec = '#e74c3c', 'š“', 'NOT RECOMMENDED'
elif 'Marginal' in label:
color, emoji, rec = '#f39c12', 'š”', 'PROCEED WITH CAUTION'
else:
color, emoji, rec = '#27ae60', 'š¢', 'GOOD OPPORTUNITY'
return f"""
{emoji} {label}
Confidence: {conf:.1%}
{rec}
Prediction based on data up to: {date.strftime('%Y-%m-%d')}
Window: {window} days
"""
def create_confidence_chart(probs):
categories = ['Unprofitable', 'Marginal', 'Profitable']
values = [probs['unprofitable'], probs['marginal'], probs['profitable']]
colors = ['#e74c3c', '#f39c12', '#27ae60']
fig = go.Figure()
fig.add_trace(go.Bar(x=categories, y=values, marker_color=colors, text=[f'{v:.1%}' for v in values], textposition='auto'))
fig.update_layout(title='Prediction Confidence', yaxis_title='Probability', yaxis=dict(range=[0, 1], tickformat='.0%'),
template='plotly_dark', height=300, margin=dict(l=0, r=0, t=40, b=0))
return fig
def create_price_chart(df, window):
# Show more context if available
display_days = min(len(df), window * 2)
df_display = df.tail(display_days)
fig = go.Figure()
fig.add_trace(go.Scatter(x=df_display['date'], y=df_display['bitcoin_price'], mode='lines', name='Bitcoin Price', line=dict(color='#F7931A', width=2)))
fig.update_layout(title=f'Bitcoin Price ({len(df_display)} Days)', xaxis_title='Date', yaxis_title='Price (USD)',
template='plotly_dark', height=300, margin=dict(l=0, r=0, t=40, b=0))
return fig
def create_interface():
with gr.Blocks(title="MineROI-Net") as app:
gr.Markdown("# šŖ MineROI-Net: Bitcoin Mining Hardware ROI Predictor")
gr.Markdown(
"Uses the **latest 30 days** of Bitcoin network data from blockchain.com "
"to classify your miner as Unprofitable / Marginal / Profitable."
)
with gr.Row():
# ---- Left: inputs ----
with gr.Column(scale=1):
gr.Markdown("### Configuration")
machine_price = gr.Number(
label="Machine price (USD)",
value=2500.0,
precision=2,
)
machine_hashrate = gr.Number(
label="Machine hashrate (TH/s)",
value=100.0,
precision=2,
)
machine_power = gr.Number(
label="Power (W)",
value=3000.0,
precision=1,
)
machine_efficiency = gr.Number(
label="Efficiency (W/TH)",
value=30.0,
precision=2,
)
machine_release_date = gr.Textbox(
label="Release date (YYYY-MM-DD)",
value="2020-05-01",
placeholder="e.g. 2020-05-01",
lines=1,
scale=1,
container=True,
show_label=True,
)
electricity_rate = gr.Number(
label="Electricity rate (USD/kWh)",
value=0.07, # neutral default
precision=4,
)
btn = gr.Button("š® Predict ROI", variant="primary", size="lg")
gr.Markdown(
"""
### About
- š“ **Unprofitable** (ROI ⤠0)
- š” **Marginal** (0 < ROI < 1)
- š¢ **Profitable** (ROI ā„ 1)
**Model:** trained on 30-day windows of Bitcoin network and miner features.
**Live mode:** whenever you click *Predict*, the app pulls the latest blockchain data.
"""
)
# ---- Right: outputs ----
with gr.Column(scale=2):
gr.Markdown("### Results")
miner_info = gr.HTML()
prediction = gr.HTML()
with gr.Row():
conf_plot = gr.Plot()
price_plot = gr.Plot()
# Connect button to prediction function
btn.click(
fn=predict_roi,
inputs=[
machine_price,
machine_hashrate,
machine_power,
machine_efficiency,
electricity_rate,
machine_release_date
],
outputs=[miner_info, prediction, conf_plot, price_plot],
)
return app
if __name__ == "__main__":
# Initialize app (loads complete data into memory)
init_app()
# Launch
app = create_interface()
app.launch()
# app.launch(server_name="0.0.0.0", server_port=7860, share=True)