Merge branch 'feature_20240513_zhaohaipeng_ad_union_thompson' into test

ad-engine-commons/pom.xml

@@ -25,7 +25,7 @@
         <dependency>
             <groupId>com.tzld.piaoquan</groupId>
             <artifactId>recommend-feature-client</artifactId>
-            <version>1.1.17</version>
+            <version>1.1.18</version>
         </dependency>
         <dependency>
             <groupId>com.tzld.piaoquan</groupId>

ad-engine-commons/src/main/java/com/tzld/piaoquan/ad/engine/commons/score/ScorerUtils.java

@@ -25,7 +25,7 @@ public final class ScorerUtils {
 
     public static String BREAK_CONFIG = "feeds_score_config_break.conf";
     public static String SHARE0_CONFIG = "feeds_score_config_share0.conf";
-
+    public static String UNION_THOMPSON_CONF = "union_score_config_thompson";
 
     public static void warmUp() {
         log.info("scorer warm up ");
@@ -33,6 +33,7 @@ public final class ScorerUtils {
         ScorerUtils.init(THOMPSON_CONF);
         ScorerUtils.init(BREAK_CONFIG);
         ScorerUtils.init(SHARE0_CONFIG);
+        ScorerUtils.init(UNION_THOMPSON_CONF);
 
     }
 

ad-engine-commons/src/main/java/com/tzld/piaoquan/ad/engine/commons/score/UnionThompsonSamplingScore.java

@@ -0,0 +1,149 @@
+package com.tzld.piaoquan.ad.engine.commons.score;
+
+import com.tzld.piaoquan.ad.engine.commons.score.model.UnionThompsonSamplingModel;
+import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRankItem;
+import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRequestContext;
+import com.tzld.piaoquan.recommend.feature.domain.ad.base.UserAdFeature;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.*;
+
+public class UnionThompsonSamplingScore extends BaseThompsonSamplingScorer {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(UnionThompsonSamplingModel.class);
+    private static final int LOCAL_TIME_OUT = 150;
+    private static final ExecutorService executorService = Executors.newFixedThreadPool(8);
+
+    public UnionThompsonSamplingScore(ScorerConfigInfo scorerConfigInfo) {
+        super(scorerConfigInfo);
+    }
+
+    @Override
+    public void loadModel() {
+        doLoadModel(UnionThompsonSamplingModel.class);
+    }
+
+    @Override
+    public List<AdRankItem> scoring(ScoreParam param, UserAdFeature userFeature, List<AdRankItem> rankItems) {
+        if (CollectionUtils.isEmpty(rankItems)) {
+            return rankItems;
+        }
+        long startTime = System.currentTimeMillis();
+
+        List<AdRankItem> result = rankByJava(rankItems, param.getRequestContext(), userFeature);
+
+        LOGGER.debug("union thompson sampling ctr ranker time java items size={}, time={} ",
+                result.size(), System.currentTimeMillis() - startTime);
+
+        return result;
+    }
+
+    private List<AdRankItem> rankByJava(final List<AdRankItem> items,
+                                        final AdRequestContext requestContext,
+                                        final UserAdFeature user) {
+        long startTime = System.currentTimeMillis();
+        UnionThompsonSamplingModel model = (UnionThompsonSamplingModel) this.getModel();
+        LOGGER.debug("model size: [{}]", model.getModelSize());
+
+        // 所有都参与打分，按照ctr排序
+        multipleCtrScore(items, model);
+
+        // debug log
+        if (LOGGER.isDebugEnabled()) {
+            for (AdRankItem item : items) {
+                LOGGER.debug("after enter feeds model predict ctr score [{}] [{}]", item, item.getScore());
+            }
+        }
+
+        LOGGER.debug("union thompson ranker java execute time: [{}]", System.currentTimeMillis() - startTime);
+        LOGGER.debug("[union thompson ranker time java] items size={}, cost={} ",
+                items.size(), System.currentTimeMillis() - startTime);
+        return items;
+    }
+
+
+    /**
+     * 计算 predict ecpm
+     */
+    public void calcScore(final UnionThompsonSamplingModel model, final AdRankItem item) {
+
+        double pctr = 0.0;
+        double pcvr = 0.0;
+        double ecpm = 0.0;
+
+        try {
+            pctr = model.ctrScore(item);
+            pcvr = model.cvrScore(item);
+            ecpm = item.getCpa() * pctr * pcvr;
+        } catch (
+                Exception e) {
+            LOGGER.error("score error for doc={} exception={}",
+                    item.getAdId(), ExceptionUtils.getFullStackTrace(e));
+        }
+        item.setTf_ctr(pctr);
+        item.setTf_cvr(pcvr);
+        item.setEcpm1(ecpm);
+        item.setScore(ecpm);
+    }
+
+
+    /**
+     * 并行打分 ecpm
+     *
+     * @param items
+     * @param model
+     */
+    private void multipleCtrScore(final List<AdRankItem> items,
+                                  final UnionThompsonSamplingModel model) {
+
+        List<Callable<Object>> calls = new ArrayList<Callable<Object>>();
+        for (int index = 0; index < items.size(); index++) {
+            final int fIndex = index;
+            items.get(fIndex).setScore(0.0);   // 设置为原始值为0
+            calls.add(new Callable<Object>() {
+                @Override
+                public Object call() {
+                    try {
+                        calcScore(model, items.get(fIndex));
+                    } catch (
+                            Exception e) {
+                        LOGGER.error("thompson exception: [{}] [{}]", items.get(fIndex).adId, ExceptionUtils.getFullStackTrace(e));
+                    }
+                    return new Object();
+                }
+            });
+        }
+
+        List<Future<Object>> futures = null;
+        try {
+            futures = executorService.invokeAll(calls, LOCAL_TIME_OUT, TimeUnit.MILLISECONDS);
+        } catch (
+                InterruptedException e) {
+            LOGGER.error("execute invoke fail: {}", ExceptionUtils.getFullStackTrace(e));
+        }
+
+        // 等待所有请求的结果返回, 超时也返回
+        int cancel = 0;
+        if (futures != null) {
+            for (Future<Object> future : futures) {
+                try {
+                    if (!future.isDone() || future.isCancelled() || future.get() == null) {
+                        cancel++;
+                    }
+                } catch (
+                        InterruptedException e) {
+                    LOGGER.error("InterruptedException {}", ExceptionUtils.getFullStackTrace(e));
+                } catch (
+                        ExecutionException e) {
+                    LOGGER.error("ExecutionException {}", ExceptionUtils.getFullStackTrace(e));
+                }
+            }
+        }
+        LOGGER.debug("ecpm Score {}, Total: {}, Cancel: {}", new Object[]{items.size(), cancel});
+    }
+}

ad-engine-commons/src/main/java/com/tzld/piaoquan/ad/engine/commons/score/model/UnionThompsonSamplingModel.java

@@ -0,0 +1,93 @@
+package com.tzld.piaoquan.ad.engine.commons.score.model;
+
+import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdActionFeature;
+import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRankItem;
+import org.apache.commons.math3.distribution.BetaDistribution;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.math.BigDecimal;
+import java.util.HashMap;
+import java.util.Map;
+
+public class UnionThompsonSamplingModel extends Model {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(UnionThompsonSamplingModel.class);
+
+    private Map<Long, Map<Long, AdActionFeature>> featureCache = new HashMap<>();
+
+    @Override
+    public int getModelSize() {
+        return featureCache.size();
+    }
+
+    @Override
+    public boolean loadFromStream(InputStreamReader in) throws Exception {
+        Map<Long, Map<Long, AdActionFeature>> initModel = new HashMap<>();
+        try (BufferedReader reader = new BufferedReader(in)) {
+
+            String line = null;
+            int size = 0;
+            while ((line = reader.readLine()) != null) {
+                String[] items = line.split("\t");
+                if (items.length < 5) {
+                    continue;
+                }
+                this.putFeature(
+                        initModel,
+                        new BigDecimal(items[0].trim()).longValue(),
+                        new BigDecimal(items[1].trim()).longValue(),
+                        new BigDecimal(items[2].trim()).doubleValue(),
+                        new BigDecimal(items[3].trim()).doubleValue(),
+                        new BigDecimal(items[4].trim()).doubleValue()
+                );
+                size++;
+            }
+            this.featureCache = initModel;
+            LOGGER.info("[UNION THOMPSON SAMPLING MODEL LOAD] mode load over, size {}", size);
+        } catch (
+                Exception e) {
+            LOGGER.info("[UNION THOMPSON SAMPLING MODEL LOAD] mode load error ", e);
+        } finally {
+            in.close();
+        }
+
+        return true;
+    }
+
+    private void putFeature(Map<Long, Map<Long, AdActionFeature>> modelCache, Long mediaId, Long creativeId, Double view, Double click, Double conversion) {
+        Map<Long, AdActionFeature> mediaCache = modelCache.computeIfAbsent(mediaId, i -> new HashMap<>());
+        AdActionFeature adActionFeature = new AdActionFeature();
+        adActionFeature.setAdView(view);
+        adActionFeature.setAdClick(click);
+        adActionFeature.setAdConversion(conversion);
+        mediaCache.put(creativeId, adActionFeature);
+
+    }
+
+    public double ctrScore(AdRankItem item) {
+        Map<Long, AdActionFeature> creativeMap = this.featureCache.getOrDefault(item.getMediaId(), new HashMap<>());
+        AdActionFeature adActionFeature = creativeMap.getOrDefault(item.getAdId(), new AdActionFeature());
+
+        int alpha_cvr = (int) adActionFeature.getAdClick() + 10;
+        int beta_cvr = 100 + (int) adActionFeature.getAdView() - (int) adActionFeature.getAdClick();
+        return this.betaSampler(alpha_cvr, beta_cvr);
+    }
+
+    public double cvrScore(AdRankItem item) {
+        Map<Long, AdActionFeature> creativeMap = this.featureCache.getOrDefault(item.getMediaId(), new HashMap<>());
+        AdActionFeature adActionFeature = creativeMap.getOrDefault(item.getAdId(), new AdActionFeature());
+
+        int alpha_cvr = (int) adActionFeature.getAdConversion() + 10;
+        int beta_cvr = 100 + (int) adActionFeature.getAdClick() - (int) adActionFeature.getAdConversion();
+        return this.betaSampler(alpha_cvr, beta_cvr);
+    }
+
+    public double betaSampler(double alpha, double beta) {
+        BetaDistribution betaSample = new BetaDistribution(alpha, beta);
+        return betaSample.sample();
+
+    }
+}

ad-engine-server/src/main/java/com/tzld/piaoquan/ad/engine/server/controller/AdRecommendController.java

@@ -4,6 +4,7 @@ import com.tzld.piaoquan.ad.engine.service.score.RankService;
 import com.tzld.piaoquan.ad.engine.service.score.dto.AdPlatformCreativeDTO;
 import com.tzld.piaoquan.ad.engine.service.score.param.BidRankRecommendRequestParam;
 import com.tzld.piaoquan.ad.engine.service.score.param.RankRecommendRequestParam;
+import com.tzld.piaoquan.ad.engine.service.score.param.UnionRankRecommendRequestParam;
 import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRankItem;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -26,36 +27,53 @@ public class AdRecommendController {
     RankService rankService;
 
     @RequestMapping("/top1/basic")
-    public Map<String,Object> adRecommendTop1Basic(@RequestBody RankRecommendRequestParam request){
+    public Map<String, Object> adRecommendTop1Basic(@RequestBody RankRecommendRequestParam request) {
         AdRankItem rankResult = rankService.adItemRank(request);
-        HashMap map =new HashMap();
-        map.put("code","0");
-        map.put("msg","success");
-        HashMap contentMap=new HashMap<>();
+        HashMap map = new HashMap();
+        map.put("code", "0");
+        map.put("msg", "success");
+        HashMap contentMap = new HashMap<>();
         contentMap.put("adId", rankResult.getAdId());
         contentMap.put("adScore", rankResult.getScore());
-        map.put("content",contentMap);
+        map.put("content", contentMap);
         return map;
     }
 
     @RequestMapping("/top1/bid/basic")
-    public Map<String,Object> adRecommendTop1BidBasic(@RequestBody BidRankRecommendRequestParam request){
+    public Map<String, Object> adRecommendTop1BidBasic(@RequestBody BidRankRecommendRequestParam request) {
         AdPlatformCreativeDTO rankResult = rankService.adBidRank(request);
-        HashMap map =new HashMap();
-        map.put("code","0");
-        map.put("msg","success");
-        map.put("content",rankResult);
+        HashMap map = new HashMap();
+        map.put("code", "0");
+        map.put("msg", "success");
+        map.put("content", rankResult);
         return map;
     }
 
     @RequestMapping("/top1/bid/new/pid")
-    public Map<String,Object> adRecommendTop1BidNewPid(@RequestBody BidRankRecommendRequestParam request){
+    public Map<String, Object> adRecommendTop1BidNewPid(@RequestBody BidRankRecommendRequestParam request) {
         AdPlatformCreativeDTO rankResult = rankService.adBidRankNewPid(request);
-        HashMap map =new HashMap();
-        map.put("code","0");
-        map.put("msg","success");
-        map.put("content",rankResult);
+        HashMap map = new HashMap();
+        map.put("code", "0");
+        map.put("msg", "success");
+        map.put("content", rankResult);
         return map;
     }
 
+    @RequestMapping("/union/top1/basic")
+    public Map<String, Object> unionAdRecommendTop1Basic(@RequestBody UnionRankRecommendRequestParam request) {
+        Map<String, Object> result = new HashMap<>();
+        AdRankItem adRankItem = rankService.unionAdItemRank(request);
+
+        Map<String, Object> content = new HashMap<>();
+        content.put("adId", adRankItem.getAdId());
+        content.put("adScore", adRankItem.getScore());
+        content.put("mediaId", adRankItem.getMediaId());
+
+        result.put("code", "0");
+        result.put("msg", "success");
+        result.put("content", content);
+
+        return result;
+    }
+
 }

ad-engine-service/src/main/java/com/tzld/piaoquan/ad/engine/service/score/RankService.java

@@ -4,6 +4,7 @@ import com.tzld.piaoquan.ad.engine.commons.score.ScoreParam;
 import com.tzld.piaoquan.ad.engine.service.score.dto.AdPlatformCreativeDTO;
 import com.tzld.piaoquan.ad.engine.service.score.param.BidRankRecommendRequestParam;
 import com.tzld.piaoquan.ad.engine.service.score.param.RankRecommendRequestParam;
+import com.tzld.piaoquan.ad.engine.service.score.param.UnionRankRecommendRequestParam;
 import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRankItem;
 import com.tzld.piaoquan.recommend.feature.domain.ad.base.UserAdFeature;
 
@@ -17,4 +18,6 @@ public interface RankService {
     AdPlatformCreativeDTO adBidRankNewPid(BidRankRecommendRequestParam request);
 
     List<AdRankItem> rank (ScoreParam param, UserAdFeature userAdFeature, List<AdRankItem> rankItems, String configFile);
+
+    AdRankItem unionAdItemRank(UnionRankRecommendRequestParam request);
 }

ad-engine-service/src/main/java/com/tzld/piaoquan/ad/engine/service/score/dto/AdUnionCreativeDTO.java

@@ -0,0 +1,16 @@
+package com.tzld.piaoquan.ad.engine.service.score.dto;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+@Data
+@ToString
+@SuperBuilder
+@EqualsAndHashCode(callSuper = true)
+public class AdUnionCreativeDTO extends AdPlatformCreativeDTO {
+
+    private Long mediaId;
+
+}

ad-engine-service/src/main/java/com/tzld/piaoquan/ad/engine/service/score/impl/RankServiceImpl.java

@@ -6,7 +6,9 @@ import com.tzld.piaoquan.ad.engine.service.score.container.PidLambdaContainer;
 import com.tzld.piaoquan.ad.engine.service.score.container.PidLambdaForCpcContainer;
 import com.tzld.piaoquan.ad.engine.service.score.container.PidLambdaV2Container;
 import com.tzld.piaoquan.ad.engine.service.score.dto.AdPlatformCreativeDTO;
+import com.tzld.piaoquan.ad.engine.service.score.dto.AdUnionCreativeDTO;
 import com.tzld.piaoquan.ad.engine.service.score.param.BidRankRecommendRequestParam;
+import com.tzld.piaoquan.ad.engine.service.score.param.UnionRankRecommendRequestParam;
 import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdItemFeature;
 import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRankItem;
 import com.tzld.piaoquan.recommend.feature.domain.ad.base.UserAdFeature;
@@ -27,6 +29,7 @@ import org.springframework.util.CollectionUtils;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 @Service
@@ -478,4 +481,59 @@ public class RankServiceImpl implements RankService {
         return rankResult;
     }
 
+    @Override
+    public AdRankItem unionAdItemRank(UnionRankRecommendRequestParam request) {
+
+        ScoreParam param= RequestConvert.requestConvert(request);
+
+        LocalDateTime currentTime = LocalDateTime.now();
+        int currentHour = currentTime.getHour();
+        int dayOfWeek = currentTime.getDayOfWeek().getValue();
+        param.getRequestContext().setHour(currentHour+"");
+        param.getRequestContext().setWeek(dayOfWeek+"");
+        param.getRequestContext().setRegion(request.getRegion().replace("省",""));
+        param.getRequestContext().setCity(request.getCity().replace("市",""));
+        param.getRequestContext().setDay(currentTime.format(dateFormatter));
+
+        List<AdRankItem> rankResult;
+        List<AdRankItem> rankItems = new LinkedList<>();
+        for (AdPlatformCreativeDTO dto : request.getCreativeList()) {
+            AdRankItem item = new AdRankItem();
+            item.setBid1(dto.getBid1());
+            item.setBid2(dto.getBid2());
+            item.setAdId(dto.getCreativeId());
+            item.setItemFeature(new AdItemFeature());
+            rankItems.add(item);
+        }
+        rankResult = rank(param, null, rankItems, ScorerUtils.UNION_THOMPSON_CONF);
+
+        AdRankItem topItem=rankResult.get(0);
+
+        Optional<AdUnionCreativeDTO> first = request.getCreativeList().stream()
+                .filter(dto -> Objects.equals(dto.getCreativeId(), topItem.getAdId()))
+                .findFirst();
+        if (first.isPresent()) {
+            AdUnionCreativeDTO dto = first.get();
+            JSONObject object = new JSONObject();
+            object.put("mid", request.getMid());
+            object.put("adid", dto.getCreativeId());
+            object.put("creativeCode", dto.getCreativeCode());
+            object.put("type", topItem.getScore_type());
+            object.put("pctr", topItem.getCtr());
+            object.put("pcvr", topItem.getCvr());
+            object.put("pidLambda", topItem.getPidLambda());
+            // 临时加入供pid v2使用
+            object.put("creativeId", dto.getCreativeId());
+            // CPA还原
+            object.put("cpa", dto.getCpa() * dto.getBid1());
+            object.put("oCpa", dto.getCpa());
+            object.put("dataTime", currentTime.format(timeFormatter));
+            log.info("svc=pid_log obj={}", JSONObject.toJSONString(object));
+            object.put("lrsamples", topItem.getLrSampleString());
+            log.info("svc=adBidRank {}", JSONObject.toJSONString(object));
+        }
+
+        return topItem;
+    }
+
 }

ad-engine-service/src/main/java/com/tzld/piaoquan/ad/engine/service/score/impl/RankServiceThompsonImpl.java

@@ -9,6 +9,7 @@ import com.tzld.piaoquan.ad.engine.service.score.convert.RequestConvert;
 import com.tzld.piaoquan.ad.engine.service.score.dto.AdPlatformCreativeDTO;
 import com.tzld.piaoquan.ad.engine.service.score.param.BidRankRecommendRequestParam;
 import com.tzld.piaoquan.ad.engine.service.score.param.RankRecommendRequestParam;
+import com.tzld.piaoquan.ad.engine.service.score.param.UnionRankRecommendRequestParam;
 import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRankItem;
 import com.tzld.piaoquan.recommend.feature.domain.ad.base.UserAdFeature;
 import org.slf4j.Logger;
@@ -83,4 +84,9 @@ public class RankServiceThompsonImpl implements RankService {
         return rankResult;
     }
 
+    @Override
+    public AdRankItem unionAdItemRank(UnionRankRecommendRequestParam request) {
+        return null;
+    }
+
 }

ad-engine-service/src/main/java/com/tzld/piaoquan/ad/engine/service/score/param/UnionRankRecommendRequestParam.java

@@ -0,0 +1,13 @@
+package com.tzld.piaoquan.ad.engine.service.score.param;
+
+import com.tzld.piaoquan.ad.engine.service.score.dto.AdUnionCreativeDTO;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.util.List;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class UnionRankRecommendRequestParam extends RecommendRequestParam{
+    List<AdUnionCreativeDTO> creativeList;
+}