AI-based Ad Performance Tracker Python, API
👤 Sharing: AI
```python
import requests
import json
import time
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error
import joblib
# --- API Configuration (Replace with your actual API details) ---
API_ENDPOINT = "YOUR_API_ENDPOINT_HERE" # Example: "https://api.example.com/ads"
API_KEY = "YOUR_API_KEY_HERE" # Your API authentication key
DATA_REFRESH_INTERVAL = 3600 # Seconds between data refreshes (e.g., 1 hour)
MODEL_REFRESH_INTERVAL = 86400 # Seconds between model refreshes (e.g., 1 day)
MODEL_FILE = "ad_performance_model.joblib" # Filename to save the trained model
HISTORICAL_DATA_FILE = "historical_ad_data.csv"
# --- Data Collection Functions ---
def fetch_ad_data_from_api(endpoint, api_key):
"""
Fetches ad performance data from the API.
Args:
endpoint (str): The API endpoint URL.
api_key (str): The API key for authentication.
Returns:
list: A list of dictionaries, where each dictionary represents an ad's performance data.
Returns None if there is an API error.
"""
try:
headers = {"Authorization": f"Bearer {api_key}"} # Or whatever auth method is appropriate
response = requests.get(endpoint, headers=headers)
response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
data = response.json() # Assumes API returns JSON data
if not isinstance(data, list):
print(f"API response is not a list. Check the API documentation.")
return None
return data
except requests.exceptions.RequestException as e:
print(f"API request failed: {e}")
return None
except json.JSONDecodeError:
print("Failed to decode JSON from API response. Check API response format.")
return None
def save_data_to_csv(data, filename):
"""
Saves ad performance data to a CSV file.
Args:
data (list): A list of dictionaries containing ad data.
filename (str): The name of the CSV file to save to.
"""
try:
df = pd.DataFrame(data)
df.to_csv(filename, index=False)
print(f"Data saved to {filename}")
except Exception as e:
print(f"Error saving data to CSV: {e}")
def load_data_from_csv(filename):
"""
Loads ad performance data from a CSV file.
Args:
filename (str): The name of the CSV file to load from.
Returns:
pandas.DataFrame: A Pandas DataFrame containing the ad data.
Returns an empty DataFrame if the file doesn't exist or an error occurs.
"""
try:
return pd.read_csv(filename)
except FileNotFoundError:
print(f"File not found: {filename}. Returning an empty DataFrame.")
return pd.DataFrame()
except Exception as e:
print(f"Error loading data from CSV: {e}. Returning an empty DataFrame.")
return pd.DataFrame()
# --- Model Training & Prediction Functions ---
def train_model(data, features, target):
"""
Trains a linear regression model on the ad performance data.
Args:
data (pandas.DataFrame): A Pandas DataFrame containing the ad data.
features (list): A list of column names to use as features.
target (str): The column name of the target variable (e.g., 'conversions').
Returns:
sklearn.linear_model.LinearRegression: The trained linear regression model.
"""
try:
X = data[features]
y = data[target]
# Handle missing values (imputation) - A more sophisticated approach might be better.
X = X.fillna(X.mean())
y = y.fillna(y.mean())
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) # Train/test split
model = LinearRegression()
model.fit(X_train, y_train)
# Evaluate the model on the test set
y_pred = model.predict(X_test)
rmse = mean_squared_error(y_test, y_pred, squared=False) #Root Mean Squared Error
print(f"Model RMSE: {rmse}")
return model
except Exception as e:
print(f"Error training model: {e}")
return None
def predict_performance(model, data, features):
"""
Predicts ad performance using the trained model.
Args:
model (sklearn.linear_model.LinearRegression): The trained model.
data (pandas.DataFrame): A Pandas DataFrame containing the ad data for which to make predictions.
features (list): A list of column names to use as features.
Returns:
pandas.Series: A Pandas Series containing the predicted performance values.
"""
try:
X = data[features]
X = X.fillna(X.mean()) # Impute missing values in the prediction data
predictions = model.predict(X)
return pd.Series(predictions, index=data.index) # Ensure index is preserved
except Exception as e:
print(f"Error predicting performance: {e}")
return None
def save_model(model, filename):
"""
Saves the trained model to a file.
Args:
model (sklearn.linear_model.LinearRegression): The trained model.
filename (str): The name of the file to save the model to.
"""
try:
joblib.dump(model, filename)
print(f"Model saved to {filename}")
except Exception as e:
print(f"Error saving model: {e}")
def load_model(filename):
"""
Loads a trained model from a file.
Args:
filename (str): The name of the file to load the model from.
Returns:
sklearn.linear_model.LinearRegression: The loaded model.
Returns None if the file doesn't exist or an error occurs.
"""
try:
return joblib.load(filename)
except FileNotFoundError:
print(f"Model file not found: {filename}. Returning None.")
return None
except Exception as e:
print(f"Error loading model: {e}. Returning None.")
return None
# --- Main Loop ---
def main():
"""
Main function to run the AI-based ad performance tracker.
"""
last_data_refresh = 0
last_model_refresh = 0
model = load_model(MODEL_FILE) # Load the model at startup.
while True:
current_time = time.time()
# Data Refresh
if current_time - last_data_refresh >= DATA_REFRESH_INTERVAL:
print("Refreshing ad data...")
new_data = fetch_ad_data_from_api(API_ENDPOINT, API_KEY)
if new_data:
# Load historical data
historical_data = load_data_from_csv(HISTORICAL_DATA_FILE)
# Convert new_data to DataFrame
new_data_df = pd.DataFrame(new_data)
# Concatenate historical and new data
all_data = pd.concat([historical_data, new_data_df], ignore_index=True)
# Save updated data to CSV
save_data_to_csv(all_data.to_dict('records'), HISTORICAL_DATA_FILE)
last_data_refresh = current_time
else:
print("Failed to fetch new data. Using existing data.")
# Model Refresh
if current_time - last_model_refresh >= MODEL_REFRESH_INTERVAL:
print("Refreshing model...")
# Load data for training
training_data = load_data_from_csv(HISTORICAL_DATA_FILE)
if not training_data.empty:
# Define features and target variable (adjust these based on your data)
features = ['impressions', 'clicks', 'spend', 'reach'] #Example features
target = 'conversions' #Example target
# Train the model
model = train_model(training_data, features, target)
if model:
# Save the trained model
save_model(model, MODEL_FILE)
last_model_refresh = current_time
else:
print("Failed to train new model. Using the existing model (if any).")
else:
print("No training data available. Skipping model refresh.")
# Prediction Example (You would integrate this with your application)
if model is not None:
# Load the latest ad data for prediction
prediction_data = load_data_from_csv(HISTORICAL_DATA_FILE).tail(5) # Example: predict the last 5 entries
features = ['impressions', 'clicks', 'spend', 'reach']
if not prediction_data.empty:
# Make predictions
predictions = predict_performance(model, prediction_data, features)
if predictions is not None:
print("Ad Performance Predictions:")
print(predictions)
else:
print("Failed to generate predictions.")
else:
print("No data available for predictions.")
# Wait before the next iteration
time.sleep(60) # Check every minute
if __name__ == "__main__":
main()
```
👁️ Viewed: 19
Comments