Skip to content
Browse files

Merge pull request #4 from visualmotive/master

Script for Mixpanel A/B test integration
  • Loading branch information...
2 parents e80ee50 + 88fc994 commit 0a7fcda05d4573886132ef24e89561bd0a1c6874 @showard showard committed
Showing with 162 additions and 0 deletions.
  1. +162 −0 tools/mixpanel_experiment.py
View
162 tools/mixpanel_experiment.py
@@ -0,0 +1,162 @@
+#! /usr/bin/env python
+#
+# Mixpanel, Inc. -- http://mixpanel.com/
+#
+# Python API client library to consume mixpanel.com analytics data.
+#
+# Modified by Chris Mueller at Thumbtack for A/B tests
+
+import hashlib
+import urllib
+import time
+import datetime
+import sys
+import webbrowser
+
+try:
+ import json
+except ImportError:
+ import simplejson as json
+
+API_KEY = 'xxxxxxxxxxxx'
+API_SECRET = 'xxxxxxxxxxxx'
+
+class Mixpanel(object):
+
+ ENDPOINT = 'http://mixpanel.com/api'
+ VERSION = '2.0'
+
+ def __init__(self, api_key, api_secret):
+ self.api_key = api_key
+ self.api_secret = api_secret
+
+ def request(self, methods, params, format='json'):
+ """
+ methods - List of methods to be joined, e.g. ['events', 'properties', 'values']
+ will give us http://mixpanel.com/api/2.0/events/properties/values/
+ params - Extra parameters associated with method
+ """
+ params['api_key'] = self.api_key
+ params['expire'] = int(time.time()) + 600 # Grant this request 10 minutes.
+ params['format'] = format
+ if 'sig' in params: del params['sig']
+ params['sig'] = self.hash_args(params)
+
+ request_url = '/'.join([self.ENDPOINT, str(self.VERSION)] + methods) + '/?' + self.unicode_urlencode(params)
+
+ request = urllib.urlopen(request_url)
+ data = request.read()
+
+ return json.loads(data)
+
+ def unicode_urlencode(self, params):
+ """
+ Convert lists to JSON encoded strings, and correctly handle any
+ unicode URL parameters.
+ """
+ if isinstance(params, dict):
+ params = params.items()
+ for i, param in enumerate(params):
+ if isinstance(param[1], list):
+ params[i] = (param[0], json.dumps(param[1]),)
+
+ return urllib.urlencode(
+ [(k, isinstance(v, unicode) and v.encode('utf-8') or v) for k, v in params]
+ )
+
+ def hash_args(self, args, secret=None):
+ """
+ Hashes arguments by joining key=value pairs, appending a secret, and
+ then taking the MD5 hex digest.
+ """
+ for a in args:
+ if isinstance(args[a], list): args[a] = json.dumps(args[a])
+
+ args_joined = ''
+ for a in sorted(args.keys()):
+ if isinstance(a, unicode):
+ args_joined += a.encode('utf-8')
+ else:
+ args_joined += str(a)
+
+ args_joined += '='
+
+ if isinstance(args[a], unicode):
+ args_joined += args[a].encode('utf-8')
+ else:
+ args_joined += str(args[a])
+
+ hash = hashlib.md5(args_joined)
+
+ if secret:
+ hash.update(secret)
+ elif self.api_secret:
+ hash.update(self.api_secret)
+ return hash.hexdigest()
+
+def get_api():
+ return Mixpanel(
+ api_key = API_KEY,
+ api_secret = API_SECRET
+ )
+
+def experiment(funnel, exp, days=None):
+ """
+ funnel: a Mixpanel funnel ID
+ exp: experiment key (a Mixpanel property)
+ days: number of days back (defaults to 1 day)
+ """
+ api = get_api()
+
+ if days is None:
+ days = 0
+
+ today = datetime.datetime.now()
+ past = today - datetime.timedelta(days=int(days))
+
+ result = api.request(['funnels'], {
+ 'funnel_id': int(funnel),
+ 'from_date': past.strftime('%Y-%m-%d'),
+ 'to_date': today.strftime('%Y-%m-%d'),
+ 'on': 'properties["%s"]' % exp,
+ })
+
+ data = result['data']
+
+ final = {}
+ for day, buckets in data.iteritems():
+ for bucket_key, events in buckets.iteritems():
+ if not final.has_key(bucket_key):
+ final[bucket_key] = []
+ for i in xrange(0, len(events)):
+ if len(final[bucket_key]) < (i + 1):
+ final[bucket_key].append(0)
+ final[bucket_key][i] += events[i]['count']
+
+ abba = []
+ for key, events in final.iteritems():
+ events.reverse()
+ query = "%s=%s" % (key, ",".join([str(event) for event in events]))
+ if key == "baseline":
+ abba.insert(0, query)
+ else:
+ abba.append(query)
+
+ abba_url = "http://www.thumbtack.com/labs/abba/#" + "&".join(abba)
+
+ print abba_url
+ webbrowser.open_new_tab(abba_url)
+
+if __name__ == '__main__':
+ if len(sys.argv) < 3:
+ print "Usage: %s <funnel> <experiment> [days]" % sys.argv[0]
+ sys.exit()
+
+ funnel = sys.argv[1]
+ exp = sys.argv[2]
+
+ if len(sys.argv) > 3:
+ days = sys.argv[3]
+ experiment(funnel, exp, days)
+ else:
+ experiment(funnel, exp)

0 comments on commit 0a7fcda

Please sign in to comment.
Something went wrong with that request. Please try again.