SQLAlchemy 自動化資料函式庫操作與機器學習流程

pip install SQLAlchemy

from sqlalchemy import create_engine

# 連線到 SQLite 資料函式庫
engine = create_engine('sqlite:///example.db')

# 連線到 MySQL 資料函式庫
engine = create_engine('mysql+pymysql://user:password@localhost/dbname')

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    nickname = Column(String)

    def __repr__(self):
        return "<User(name='%s', fullname='%s', nickname='%s')>" % (
            self.name, self.fullname, self.nickname)

from sqlalchemy.orm import sessionmaker

# 建立 Session 類別
Session = sessionmaker(bind=engine)
session = Session()

# 建立新使用者
ed_user = User(name='ed', fullname='Ed Jones', nickname='edsnickname')
session.add(ed_user)
session.commit()

our_user = session.query(User).filter_by(name='ed').first()
print(our_user)

our_user.nickname = 'eddie'
session.commit()

session.delete(our_user)
session.commit()

# 查詢所有使用者
users = session.query(User).all()
for user in users:
    print(user)

# 查詢特定名稱的使用者
users = session.query(User).filter_by(name='ed').all()
for user in users:
    print(user)

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship("User", back_populates="addresses")

User.addresses = relationship("Address", order_by=Address.id, back_populates="user")

# 查詢使用者及其地址
users = session.query(User).join(Address).filter(Address.email_address=='[email protected]').all()
for user in users:
    print(user)

pip install alembic
alembic init alembic

alembic revision --autogenerate -m "create user table"
alembic upgrade head

# 自動化資料預處理
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

def preprocess_data(file_path):
    # 載入資料
    data = pd.read_csv(file_path)
    
    # 將資料分成特徵和目標
    X = data.drop('target', axis=1)
    y = data['target']
    
    # 將資料分成訓練和測試集
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
    
    # 標準化特徵
    scaler = StandardScaler()
    X_train = scaler.fit_transform(X_train)
    X_test = scaler.transform(X_test)
    
    return X_train, X_test, y_train, y_test

# 使用範例
file_path = 'data.csv'
X_train, X_test, y_train, y_test = preprocess_data(file_path)

# 自動化模型訓練
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
import joblib

def train_model(X_train, y_train):
    # 初始化模型
    model = RandomForestClassifier(n_estimators=100, random_state=42)
    
    # 訓練模型
    model.fit(X_train, y_train)
    
    # 儲存模型
    joblib.dump(model, 'model.pkl')
    
    return model

# 使用範例
model = train_model(X_train, y_train)

# 自動化超引數調優
from sklearn.model_selection import GridSearchCV

def tune_hyperparameters(model, X_train, y_train):
    # 定義引數網格
    param_grid = {
        'n_estimators': [50, 100, 200],
        'max_depth': [None, 10, 20, 30],
        'min_samples_split': [2, 5, 10]
    }
    
    # 初始化GridSearchCV
    grid_search = GridSearchCV(estimator=model, param_grid=param_grid, cv=5, scoring='accuracy')
    
    # 擬合模型
    grid_search.fit(X_train, y_train)
    
    # 儲存最佳模型
    joblib.dump(grid_search.best_estimator_, 'best_model.pkl')
    
    return grid_search.best_estimator_

# 使用範例
best_model = tune_hyperparameters(model, X_train, y_train)

# 自動化模型評估
def evaluate_model(model, X_test, y_test):
    # 在測試集上進行預測
    y_pred = model.predict(X_test)
    
    # 計算準確率
    accuracy = accuracy_score(y_test, y_pred)
    
    return accuracy

# 使用範例
accuracy = evaluate_model(best_model, X_test, y_test)
print(f'模型準確率：{accuracy}')

# 自動化模型佈署
from flask import Flask, request, jsonify
import joblib

app = Flask(__name__)

@app.route('/predict', methods=['POST'])
def predict():
    # 載入模型
    model = joblib.load('best_model.pkl')
    
    # 取得輸入資料
    data = request.get_json(force=True)
    
    # 進行預測
    prediction = model.predict([data['features']])
    
    # 傳回預測結果
    return jsonify({'prediction': prediction.tolist()})

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

# 自動化監控和維護
import requests
import pandas as pd

def monitor_model(endpoint, data_file):
    # 載入新資料
    new_data = pd.read_csv(data_file)
    
    # 使用已佈署的模型進行預測
    response = requests.post(endpoint, json={'features': new_data.iloc[0].tolist()})
    
    # 記錄預測結果
    with open('monitoring_log.txt', 'a') as f:
        f.write(f'預測結果：{response.json()["prediction"]}\n')

# 使用範例
endpoint = 'http://localhost:5000/predict'
data_file = 'new_data.csv'
monitor_model(endpoint, data_file)