fork download
  1. import os
  2. import requests
  3. import numpy as np
  4. import logging
  5. from flask import Flask, jsonify, request
  6. from transformers import pipeline
  7.  
  8. # Initialize logging
  9. logging.basicConfig(level=logging.INFO)
  10. logger = logging.getLogger(__name__)
  11.  
  12. # Flask app for interaction
  13. app = Flask(__name__)
  14.  
  15. # Load the NLP model once to improve performance
  16. try:
  17. summarizer = pipeline("summarization")
  18. except Exception as e:
  19. logger.error('Error loading summarization model: %s', e)
  20. summarizer = None
  21.  
  22. # ------------------- Online Trading Module ------------------- #
  23. def fetch_trading_data(symbol="AAPL"):
  24. """Fetch trading data using an API like Yahoo Finance."""
  25. url = f"https://q...content-available-to-author-only...o.com/v7/finance/quote?symbols={symbol}"
  26. try:
  27. response = requests.get(url)
  28. response.raise_for_status() # Raise an error for bad responses
  29. data = response.json()
  30. return data["quoteResponse"]["result"]
  31. except requests.exceptions.RequestException as e:
  32. logger.error('Error fetching trading data: %s', e)
  33. return []
  34.  
  35. def trading_strategy(data):
  36. """Simple moving average strategy."""
  37. prices = [item["regularMarketPrice"] for item in data]
  38. if len(prices) > 1 and prices[-1] > np.mean(prices[:-1]):
  39. return "BUY"
  40. else:
  41. return "SELL"
  42.  
  43. # ------------------- Cybersecurity Module ------------------- #
  44. def detect_intrusion(logs):
  45. """Simple mock intrusion detection."""
  46. threats = [log for log in logs if "malicious" in log.lower()]
  47. return threats
  48.  
  49. # ------------------- NLP Module ------------------- #
  50. def summarize_text(text):
  51. """Use Hugging Face pipeline for summarization."""
  52. if summarizer is not None:
  53. try:
  54. summary = summarizer(text, max_length=50, min_length=25, do_sample=False)
  55. return summary[0]["summary_text"]
  56. except Exception as e:
  57. logger.error('Error during text summarization: %s', e)
  58. return 'Error during summarization.'
  59. return 'Summarizer not initialized.'
  60.  
  61. # ------------------- Flask API ------------------- #
  62. @app.route('/trading/<symbol>', methods=['GET'])
  63. def trading(symbol):
  64. data = fetch_trading_data(symbol)
  65. if not data:
  66. return jsonify({"error": "Failed to fetch trading data."}), 500
  67. decision = trading_strategy(data)
  68. return jsonify({"symbol": symbol, "decision": decision})
  69.  
  70. @app.route('/cybersecurity', methods=['POST'])
  71. def cybersecurity():
  72. logs = request.json.get("logs", [])
  73. if not isinstance(logs, list):
  74. return jsonify({"error": "Logs must be provided as a list."}), 400
  75. threats = detect_intrusion(logs)
  76. return jsonify({"threats": threats})
  77.  
  78. @app.route('/summarize', methods=['POST'])
  79. def nlp_summary():
  80. text = request.json.get("text", "")
  81. if not text:
  82. return jsonify({"error": "Text must be provided."}), 400
  83. summary = summarize_text(text)
  84. return jsonify({"summary": summary})
  85.  
  86. if __name__ == "__main__":
  87. app.run(debug=True)
Success #stdin #stdout 0.02s 25956KB
stdin
Standard input is empty
stdout
import os
import requests
import numpy as np
import logging
from flask import Flask, jsonify, request
from transformers import pipeline

# Initialize logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Flask app for interaction
app = Flask(__name__)

# Load the NLP model once to improve performance
try:
    summarizer = pipeline("summarization")
except Exception as e:
    logger.error('Error loading summarization model: %s', e)
    summarizer = None

# ------------------- Online Trading Module ------------------- #
def fetch_trading_data(symbol="AAPL"):
    """Fetch trading data using an API like Yahoo Finance."""
    url = f"https://q...content-available-to-author-only...o.com/v7/finance/quote?symbols={symbol}"
    try:
        response = requests.get(url)
        response.raise_for_status()  # Raise an error for bad responses
        data = response.json()
        return data["quoteResponse"]["result"]
    except requests.exceptions.RequestException as e:
        logger.error('Error fetching trading data: %s', e)
        return []

def trading_strategy(data):
    """Simple moving average strategy."""
    prices = [item["regularMarketPrice"] for item in data]
    if len(prices) > 1 and prices[-1] > np.mean(prices[:-1]):
        return "BUY"
    else:
        return "SELL"

# ------------------- Cybersecurity Module ------------------- #
def detect_intrusion(logs):
    """Simple mock intrusion detection."""
    threats = [log for log in logs if "malicious" in log.lower()]
    return threats

# ------------------- NLP Module ------------------- #
def summarize_text(text):
    """Use Hugging Face pipeline for summarization."""
    if summarizer is not None:
        try:
            summary = summarizer(text, max_length=50, min_length=25, do_sample=False)
            return summary[0]["summary_text"]
        except Exception as e:
            logger.error('Error during text summarization: %s', e)
            return 'Error during summarization.'
    return 'Summarizer not initialized.'

# ------------------- Flask API ------------------- #
@app.route('/trading/<symbol>', methods=['GET'])
def trading(symbol):
    data = fetch_trading_data(symbol)
    if not data:
        return jsonify({"error": "Failed to fetch trading data."}), 500
    decision = trading_strategy(data)
    return jsonify({"symbol": symbol, "decision": decision})

@app.route('/cybersecurity', methods=['POST'])
def cybersecurity():
    logs = request.json.get("logs", [])
    if not isinstance(logs, list):
        return jsonify({"error": "Logs must be provided as a list."}), 400
    threats = detect_intrusion(logs)
    return jsonify({"threats": threats})

@app.route('/summarize', methods=['POST'])
def nlp_summary():
    text = request.json.get("text", "")
    if not text:
        return jsonify({"error": "Text must be provided."}), 400
    summary = summarize_text(text)
    return jsonify({"summary": summary})

if __name__ == "__main__":
    app.run(debug=True)