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)