Skip to content

Instantly share code, notes, and snippets.

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings.sentence_transformer import SentenceTransformerEmbeddings
from typing import Annotated, Dict, TypedDict
from langchain_core.messages import BaseMessage
import json
import operator
from typing import Annotated, Sequence, TypedDict
def expected_steps(df):
Q = df.drop(
['Null', 'Activation'], axis=1).drop(['Null', 'Activation'], axis=0)
I = np.identity(Q.shape[1])
N = np.linalg.inv(I - Q.to_numpy())
t = np.sum(N, axis=1)
plt.figure(figsize=(10,5))
sns.scatterplot(data=df_scatter, x='Click Activation Rate', y='Activation Rate', s=200, color='#2653de')
for line in range(0, df_scatter.shape[0]):
plt.text(df_scatter['Click Activation Rate'][line]+0.001, df_scatter['Activation Rate'][line],
df_scatter['Channel'][line], horizontalalignment='left',
size='medium', color='black', weight='semibold')
df_scatter = df_multi.copy()
df_scatter['Coverage'] = df_scatter['Channel'].map(
campaign_data.groupby('channel')['customer_id'].nunique().to_dict()
)
df_scatter['Total Clicks'] = df_scatter['Channel'].map(
journeys['path'].apply(lambda x: x[-2]).value_counts().to_dict()
)
df_multi = pd.DataFrame({
'Channel': attributions.keys(),
'Attribution style': 'Journey',
'Activations': attributions.values()
})
df_first = pd.DataFrame({
'Channel': attributions.keys(),
'Attribution style': 'First touchpoint'
})
def removal_effects(df, conversion_rate):
removal_effects_dict = {}
channels = [channel for channel in df.columns if channel not in ['Start',
'Null',
'Activation']]
for channel in channels:
removal_df = df.drop(channel, axis=1).drop(channel, axis=0)
for column in removal_df.columns:
row_sum = np.sum(list(removal_df.loc[column]))
null_pct = float(1) - row_sum
import matplotlib.pyplot as plt
import seaborn as sns
from markovchain import MarkovChain
mc = MarkovChain(trans_matrix.values, trans_matrix.index)
mc.draw()
# Function to create intermediate path strings
def transition_states(paths):
unique_channels = set(x for element in paths for x in element)
transition_states = {x + '>' + y: 0 for x in unique_channels for y in unique_channels}
for possible_state in unique_channels:
if possible_state not in ['Activation', 'Null']:
for user_path in paths:
if possible_state in user_path:
indices = [i for i, s in enumerate(user_path) if possible_state in s]
# Step 1: Convert all interactions to a list
journeys = campaign_data.groupby('customer_id')['channel'].aggregate(
lambda x: x.tolist()).reset_index()
# Step 2: Add last interaction as 1 or 0 event representing activation
activation_results = campaign_data.drop_duplicates('customer_id', keep='last')[['customer_id', 'activation']]
journeys = pd.merge(journeys, activation_results, how='left', on='customer_id')
# Step 3: Add start and end states based on whether customer activated
journeys['path'] = np.where(
import numpy as np
import pandas as pd
campaign_data = pd.read_csv("cashback_activation_data.csv")
campaign_data = campaign_data.sort_values(['customer_id', 'timestamp'],
ascending=[False, True])
campaign_data['visit_order'] = campaign_data.groupby('customer_id').cumcount() + 1