File size: 5,910 Bytes
7102613 774971c 7102613 7cb58d4 774971c c98df93 774971c 458d562 774971c e222b9e c98df93 e222b9e 7277e72 774971c b002957 7102613 7277e72 774971c b002957 e222b9e b002957 774971c b002957 71ca3e8 e222b9e c98df93 e222b9e 774971c b002957 c98df93 e222b9e b002957 e222b9e b002957 e222b9e b002957 e222b9e c98df93 e222b9e 774971c b002957 c98df93 e222b9e 774971c 7cb58d4 b002957 c98df93 e222b9e 71ca3e8 7cb58d4 33fef3a 71ca3e8 33fef3a c98df93 33fef3a 7102613 33fef3a 71ca3e8 c98df93 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 |
import os
import logging
import numpy as np
import pandas as pd
import xgboost as xgb
import requests
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
import gradio as gr
# Initialize Flask application
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL') # Default to SQLite for testing
db = SQLAlchemy(app)
# Set up logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler("app.log"), # Log to file
logging.StreamHandler() # Log to console
]
)
logger = logging.getLogger(__name__)
# Database model for storing economic indicators
class EconomicData(db.Model):
id = db.Column(db.Integer, primary_key=True)
indicator = db.Column(db.String(100))
value = db.Column(db.Float)
def initialize_db():
"""Create database tables and add dummy data if not present."""
with app.app_context():
db.create_all()
if EconomicData.query.count() == 0:
create_dummy_data()
def create_dummy_data():
"""Add dummy data to the database for initial testing."""
economic_records = [
EconomicData(indicator='GDP', value=2.5),
EconomicData(indicator='Inflation', value=1.2),
EconomicData(indicator='Unemployment Rate', value=4.5)
]
db.session.bulk_save_objects(economic_records)
db.session.commit()
@app.route('/search', methods=['POST'])
def search():
"""Search for economic indicators."""
logger.debug("Search endpoint hit.")
try:
query = request.json['query']
results = EconomicData.query.filter(EconomicData.indicator.ilike(f'%{query}%')).all()
logger.debug(f"Search results for query '{query}': {[(record.indicator, record.value) for record in results]}")
return jsonify({record.indicator: record.value for record in results})
except Exception as e:
logger.error(f"Error in search: {e}")
return jsonify({"error": str(e)}), 500
@app.route('/predict', methods=['POST'])
def predict():
"""Predict future values based on historical data."""
logger.debug("Predict endpoint hit.")
try:
data = request.json
num_predictions = data['num_predictions']
historical_data = pd.Series([100, 101, 102, 104, 107])
model = xgb.XGBRegressor()
model.fit(np.arange(len(historical_data)).reshape(-1, 1), historical_data.values)
last_days = np.array([len(historical_data)]).reshape(-1, 1)
predictions = []
for _ in range(num_predictions):
next_value = model.predict(last_days)[0]
predictions.append(next_value)
last_days = np.append(last_days, [[last_days[-1][0] + 1]], axis=0)
logger.debug(f"Predictions: {predictions}")
return jsonify({"predictions": predictions})
except Exception as e:
logger.error(f"Error in prediction: {e}")
return jsonify({"error": str(e)}), 500
@app.route('/economic_data', methods=['GET'])
def get_economic_data():
"""Retrieve all economic data records."""
logger.debug("Retrieve economic data endpoint hit.")
try:
records = EconomicData.query.all()
return jsonify({record.indicator: record.value for record in records})
except Exception as e:
logger.error(f"Error in retrieving economic data: {e}")
return jsonify({"error": str(e)}), 500
def gradio_ui():
"""Launch the Gradio interface."""
logger.debug("Launching Gradio UI.")
def call_search(query):
response = requests.post('http://localhost:5000/search', json={'query': query})
return response.json()
def call_predict(num_predictions):
response = requests.post('http://localhost:5000/predict', json={'num_predictions': num_predictions})
return response.json()
def call_economic_data():
response = requests.get('http://localhost:5000/economic_data')
return response.json()
with gr.Blocks() as demo:
gr.Markdown("# Economic DataHub")
gr.Markdown("Welcome to the interactive Economic DataHub. Explore economic indicators, make predictions, and access detailed economic data.")
with gr.Tab("Search Economic Indicators"):
search_input = gr.Textbox(label="Search for Economic Terms:", placeholder="Enter an economic term...")
search_button = gr.Button("Search")
search_output = gr.Textbox(label="Results", interactive=False)
search_button.click(call_search, inputs=search_input, outputs=search_output)
with gr.Tab("Time Series Prediction"):
num_predictions = gr.Slider(minimum=1, maximum=30, step=1, label="Number of Days to Predict")
predict_button = gr.Button("Predict")
prediction_output = gr.Textbox(label="Predictions", interactive=False)
predict_button.click(call_predict, inputs=num_predictions, outputs=prediction_output)
with gr.Tab("Retrieve Economic Data"):
retrieve_button = gr.Button("Get Economic Data")
data_output = gr.Textbox(label="Economic Data", interactive=False)
retrieve_button.click(call_economic_data, outputs=data_output)
demo.launch(server_name="0.0.0.0", server_port=7860, share=True)
if __name__ == '__main__':
initialize_db() # Ensure the database is initialized before starting the app
# Start Flask app
app.run(debug=True, host='0.0.0.0', port=5000) # Start Flask in the main thread
# Launch Gradio UI in the same thread for simplicity (no threading)
gradio_ui() |