|
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 |
|
|
|
|
|
app = Flask(__name__) |
|
app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL') |
|
db = SQLAlchemy(app) |
|
|
|
|
|
logging.basicConfig( |
|
level=logging.DEBUG, |
|
format='%(asctime)s - %(levelname)s - %(message)s', |
|
handlers=[ |
|
logging.FileHandler("app.log"), |
|
logging.StreamHandler() |
|
] |
|
) |
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
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() |
|
|
|
app.run(debug=True, host='0.0.0.0', port=5000) |
|
|
|
|
|
gradio_ui() |