projectfarhad

[leaderfazonoozi-spec]

=========================================

0) نصب و آماده‌سازی محیط

=========================================

!pip -q install openpyxl scikit-learn pandas matplotlib seaborn

import os
import re
import math
import json
import warnings
from pathlib import Path

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor, HistGradientBoostingRegressor

warnings.filterwarnings("ignore")

=========================================

1) مسیرها

=========================================

BASE_DIR = Path("/content") # اگر فایل‌ها در Drive هستند، بعداً تغییر می‌دهیم
OUT_DIR = Path("/content/output")
OUT_DIR.mkdir(parents=True, exist_ok=True)

اگر فایل‌ها در گوگل‌درایو هستند، این بخش را فعال کنید:

from google.colab import drive

drive.mount('/content/drive')

BASE_DIR = Path('/content/drive/MyDrive/your_folder_name')

OUT_DIR = BASE_DIR / "outputs"

OUT_DIR.mkdir(parents=True, exist_ok=True)

RAW_FILES = {
"competitive": BASE_DIR / "قیمت رقابتی.xlsx",
"transport": BASE_DIR / "هزینه حمل و نقل.xlsx",
"reviews": BASE_DIR / "لیست نظرات کاربران.xlsx",
"final_csv": BASE_DIR / "final_merged_dataset.csv",
"final_xlsx": BASE_DIR / "final_merged_dataset.xlsx",
}

OUTPUTS = {
"featured_dataset_csv": OUT_DIR / "final_featured_dataset.csv",
"featured_dataset_xlsx": OUT_DIR / "final_featured_dataset.xlsx",
"model_comparison_csv": OUT_DIR / "model_comparison.csv",
"predictions_csv": OUT_DIR / "best_model_predictions.csv",
"metrics_json": OUT_DIR / "best_model_metrics.json",
"plot_png": OUT_DIR / "actual_vs_predicted.png",
}

=========================================

2) توابع کمکی

=========================================

def normalize_text(value):
if pd.isna(value):
return np.nan
value = str(value).strip()
value = value.replace("ي", "ی").replace("ك", "ک")
value = re.sub(r"\s+", " ", value)
return value

def parse_numeric(value):
if pd.isna(value):
return np.nan
if isinstance(value, (int, float, np.number)):
return float(value)
text = str(value).strip()
text = text.replace(",", "").replace("٬", "").replace("ریال", "").replace("تومان", "")
text = text.replace("\u200c", " ")
text = re.sub(r"[^\d\.\-]", "", text)
if text in {"", "-", ".", "-."}:
return np.nan
try:
return float(text)
except:
return np.nan

def month_name_to_number(name):
if pd.isna(name):
return np.nan
text = normalize_text(name).lower()
mapping = {
"فروردین": 1, "اردیبهشت": 2, "خرداد": 3, "تیر": 4,
"مرداد": 5, "شهریور": 6, "مهر": 7, "آبان": 8,
"آذر": 9, "دی": 10, "بهمن": 11, "اسفند": 12,
"january": 1, "february": 2, "march": 3, "april": 4,
"may": 5, "june": 6, "july": 7, "august": 8,
"september": 9, "october": 10, "november": 11, "december": 12,
}
if text in mapping:
return mapping[text]
num = parse_numeric(text)
if pd.notna(num) and 1 <= num <= 12:
return int(num)
return np.nan

def detect_date_column(df):
candidates = [c for c in df.columns if any(k in str(c).lower() for k in ["date", "تاریخ", "زمان"])]
if candidates:
return candidates[0]
# fallback: ستونی که بیشترین تعداد قابل تبدیل به تاریخ دارد
best_col, best_score = None, -1
for col in df.columns:
score = pd.to_datetime(df[col], errors="coerce").notna().sum()
if score > best_score:
best_score = score
best_col = col
return best_col

def detect_target_column(df):
preferred = ["price", "قیمت", "قیمت فروش", "selling_price", "final_price"]
cols_lower = {str(c).strip().lower(): c for c in df.columns}
for p in preferred:
if p.lower() in cols_lower:
return cols_lower[p.lower()]
num_cols = df.select_dtypes(include=[np.number]).columns.tolist()
return num_cols[0] if num_cols else None

def mape(y_true, y_pred):
y_true = np.array(y_true, dtype=float)
y_pred = np.array(y_pred, dtype=float)
mask = y_true != 0
if mask.sum() == 0:
return np.nan
return np.mean(np.abs((y_true[mask] - y_pred[mask]) / y_true[mask])) * 100

=========================================

3) ساخت/خواندن دیتاست پایه

=========================================

def load_base_dataset():
if RAW_FILES["final_csv"].exists():
print("Loaded final_merged_dataset.csv")
df = pd.read_csv(RAW_FILES["final_csv"])
elif RAW_FILES["final_xlsx"].exists():
print("Loaded final_merged_dataset.xlsx")
df = pd.read_excel(RAW_FILES["final_xlsx"])
else:
raise FileNotFoundError(
"final_merged_dataset.csv/xlsx not found. "
"Please put raw Excel files in BASE_DIR or upload the merged dataset."
)
df.columns = [normalize_text(c) for c in df.columns]
return df

df = load_base_dataset()
print("Shape:", df.shape)
display(df.head())
print(df.columns.tolist())

=========================================

4) آماده‌سازی ستون‌ها

=========================================

df = df.copy()

تاریخ

if "date" in df.columns:
df["date"] = pd.to_datetime(df["date"], errors="coerce")
elif {"year", "month"}.issubset(df.columns):
df["year"] = pd.to_numeric(df["year"], errors="coerce")
df["month"] = pd.to_numeric(df["month"], errors="coerce")
pseudo_year = df["year"].fillna(2000).astype(int)
pseudo_month = df["month"].fillna(1).astype(int).clip(1, 12)
df["date"] = pd.to_datetime(
dict(year=np.where(pseudo_year < 1900, 2000 + (pseudo_year % 100), pseudo_year),
month=pseudo_month,
day=1),
errors="coerce"
)
else:
df["date"] = pd.RangeIndex(len(df))

هدف

if "price" not in df.columns:
target = detect_target_column(df)
if target is None:
raise ValueError("No target price column found.")
df["price"] = pd.to_numeric(df[target], errors="coerce")
else:
df["price"] = pd.to_numeric(df["price"], errors="coerce")

if "product" not in df.columns:
df["product"] = "unknown_product"
df["product"] = df["product"].apply(normalize_text)

اعدادی‌سازی ستون‌های عددی

for col in df.columns:
if col not in ["product", "date"]:
if df[col].dtype == "object":
df[col] = df[col].apply(parse_numeric)

df = df.sort_values(["product", "date"]).reset_index(drop=True)
df = df[df["price"].notna()].copy()

print("Prepared shape:", df.shape)

=========================================

5) مهندسی ویژگی

=========================================

def engineer_features(data):
data = data.copy()
data = data.sort_values(["product", "date"]).reset_index(drop=True)

data["day"] = pd.to_datetime(data["date"], errors="coerce").dt.day
data["month_num"] = pd.to_datetime(data["date"], errors="coerce").dt.month
data["dayofweek"] = pd.to_datetime(data["date"], errors="coerce").dt.dayofweek

g = data.groupby("product", group_keys=False)
data["price_lag_1"] = g["price"].shift(1)
data["price_lag_2"] = g["price"].shift(2)
data["price_lag_3"] = g["price"].shift(3)

data["price_roll_mean_3"] = g["price"].transform(lambda s: s.shift(1).rolling(3, min_periods=1).mean())
data["price_roll_std_3"] = g["price"].transform(lambda s: s.shift(1).rolling(3, min_periods=1).std())

if "competitive_price" in data.columns:
    data["price_gap_vs_competitor"] = data["price"] - data["competitive_price"]

if "transport_cost" in data.columns:
    data["price_minus_transport"] = data["price"] - data["transport_cost"]

if "purchase_quantity" in data.columns:
    data["sales_lag_1"] = g["purchase_quantity"].shift(1)

data["record_index"] = np.arange(len(data))
return data

feat_df = engineer_features(df)
print("Feature dataset shape:", feat_df.shape)

ذخیره دیتاست ویژگی‌سازی شده

feat_df.to_csv(OUTPUTS["featured_dataset_csv"], index=False, encoding="utf-8-sig")
feat_df.to_excel(OUTPUTS["featured_dataset_xlsx"], index=False)

=========================================

6) آموزش و مقایسه مدل‌ها

=========================================

def train_and_compare_models(data):
data = data.copy()

# ستون‌های غیرورودی
drop_cols = ["price"]
if "review_text" in data.columns:
    drop_cols.append("review_text")

feature_cols = [c for c in data.columns if c not in drop_cols]

# date را به ordinal تبدیل می‌کنیم
if "date" in feature_cols:
    data["date_ordinal"] = pd.to_datetime(data["date"], errors="coerce").map(lambda x: x.toordinal() if pd.notna(x) else np.nan)
    feature_cols = [c for c in feature_cols if c != "date"] + ["date_ordinal"]

X = data[feature_cols].copy()
y = data["price"].copy()

cat_cols = [c for c in X.columns if X[c].dtype == "object"]
num_cols = [c for c in X.columns if c not in cat_cols]

preprocessor = ColumnTransformer(
    transformers=[
        ("num", Pipeline([
            ("imputer", SimpleImputer(strategy="median")),
            ("scaler", StandardScaler())
        ]), num_cols),
        ("cat", Pipeline([
            ("imputer", SimpleImputer(strategy="most_frequent")),
            ("onehot", OneHotEncoder(handle_unknown="ignore"))
        ]), cat_cols),
    ]
)

models = {
    "LinearRegression": LinearRegression(),
    "RandomForest": RandomForestRegressor(n_estimators=300, max_depth=10, random_state=42, n_jobs=-1),
    "GradientBoosting": GradientBoostingRegressor(random_state=42),
    "HistGradientBoosting": HistGradientBoostingRegressor(random_state=42),
}

# split زمانی
sort_col = "date_ordinal" if "date_ordinal" in data.columns else "record_index"
ordered_idx = data.sort_values(sort_col).index
split_idx = int(len(data) * 0.8)
train_idx = ordered_idx[:split_idx]
test_idx = ordered_idx[split_idx:]

X_train, X_test = X.loc[train_idx], X.loc[test_idx]
y_train, y_test = y.loc[train_idx], y.loc[test_idx]

results = []
fitted = {}

for name, model in models.items():
    pipe = Pipeline([
        ("preprocessor", preprocessor),
        ("model", model),
    ])
    pipe.fit(X_train, y_train)
    pred = pipe.predict(X_test)

    rmse = math.sqrt(mean_squared_error(y_test, pred))
    mae = mean_absolute_error(y_test, pred)
    r2 = r2_score(y_test, pred)
    mp = mape(y_test, pred)

    results.append({
        "model": name,
        "RMSE": rmse,
        "MAE": mae,
        "R2": r2,
        "MAPE": mp
    })

    fitted[name] = pred

results_df = pd.DataFrame(results).sort_values(["RMSE", "MAE"]).reset_index(drop=True)
best_model = results_df.loc[0, "model"]

pred_df = data.loc[test_idx, ["product", "date"]].copy()
pred_df["actual_price"] = y_test.values
pred_df["predicted_price"] = fitted[best_model]

return results_df, best_model, pred_df

results_df, best_model, pred_df = train_and_compare_models(feat_df)

print("Best model:", best_model)
display(results_df)

ذخیره جدول مقایسه

results_df.to_csv(OUTPUTS["model_comparison_csv"], index=False, encoding="utf-8-sig")

ذخیره پیش‌بینی‌ها

pred_df.to_csv(OUTPUTS["predictions_csv"], index=False, encoding="utf-8-sig")

ذخیره معیار بهترین مدل

best_metrics = results_df[results_df["model"] == best_model].iloc[0].to_dict()
with open(OUTPUTS["metrics_json"], "w", encoding="utf-8") as f:
json.dump(best_metrics, f, ensure_ascii=False, indent=2)

=========================================

7) رسم نمودار Actual vs Predicted

=========================================

plot_df = pred_df.sort_values("date").copy()

plt.figure(figsize=(14, 6))
plt.plot(plot_df["date"], plot_df["actual_price"], label="Actual", linewidth=2)
plt.plot(plot_df["date"], plot_df["predicted_price"], label="Predicted", linewidth=2)

plt.title(f"Actual vs Predicted Price - Best Model: {best_model}", fontsize=14)
plt.xlabel("Date")
plt.ylabel("Price")
plt.grid(True, alpha=0.3)
plt.legend()
plt.tight_layout()
plt.savefig(OUTPUTS["plot_png"], dpi=150)
plt.show()

print("\nSaved files:")
for k, v in OUTPUTS.items():
print(f"{k}: {v}")